feat: session ID columns, session.id from rum.sessionId, severity text from number

.changeset/session-id-columns-severity-text.md

@@ -0,0 +1,21 @@
+---
+'@hyperdx/otel-collector': patch
+---
+
+feat: session ID columns, `session.id` from `rum.sessionId`, severity text
+
+The `hyperdx_sessions` seed adds materialized `rum.sessionId` and `session.id`
+columns with skip indexes, and the `otel_logs` seed adds a materialized
+`session.id` column with a skip index, so session queries no longer read the
+attribute maps. Only newly created tables get them; existing tables need an
+`ALTER TABLE ... ADD COLUMN` to benefit.
+
+Spans, logs and session replay events from older `@hyperdx/browser` SDKs, which
+carry the session id only in the `rum.sessionId` resource attribute, now also
+get the OpenTelemetry `session.id` attribute. An existing `session.id` is never
+overwritten and `rum.sessionId` is kept. If you override pipeline `processors:`
+lists in `CUSTOM_OTELCOL_CONFIG_FILE`, add `transform/session_id` to keep this.
+
+Logs that carry a severity number but no severity text, such as those from the
+new browser SDK, now get the matching level text (`info`, `error`, ...) instead
+of an empty level. Severity text set by the producer is never overwritten.

docker/otel-collector/config.yaml

@@ -117,11 +117,50 @@ processors:
           - set(log.severity_text, "info") where log.severity_number == 0
           - set(log.severity_number, SEVERITY_NUMBER_INFO) where
             log.severity_number == 0
+      - context: log
+        error_mode: ignore
+        conditions:
+          - severity_text == "" and severity_number > 0
+        statements:
+          - set(log.severity_text, "trace") where log.severity_number < 5
+          - set(log.severity_text, "debug") where log.severity_number >= 5 and
+            log.severity_number < 9
+          - set(log.severity_text, "info") where log.severity_number >= 9 and
+            log.severity_number < 13
+          - set(log.severity_text, "warn") where log.severity_number >= 13 and
+            log.severity_number < 17
+          - set(log.severity_text, "error") where log.severity_number >= 17 and
+            log.severity_number < 21
+          - set(log.severity_text, "fatal") where log.severity_number >= 21
       - context: log
         error_mode: ignore
         statements:
           # Normalize the severity_text case
           - set(log.severity_text, ConvertCase(log.severity_text, "lower"))
+  # Older @hyperdx/browser SDKs put the session id only in the
+  # `rum.sessionId` resource attribute. Copy it to the OTel `session.id`
+  # attribute so newer and older SDK data can be queried by one key. An
+  # existing `session.id` (e.g. from a JSON body merged by `transform`) is
+  # never overwritten, and `rum.sessionId` is kept.
+  transform/session_id:
+    trace_statements:
+      - context: span
+        error_mode: ignore
+        statements:
+          - set(span.attributes["session.id"],
+            resource.attributes["rum.sessionId"]) where
+            span.attributes["session.id"] == nil and
+            IsString(resource.attributes["rum.sessionId"]) and
+            Len(resource.attributes["rum.sessionId"]) > 0
+    log_statements:
+      - context: log
+        error_mode: ignore
+        statements:
+          - set(log.attributes["session.id"],
+            resource.attributes["rum.sessionId"]) where
+            log.attributes["session.id"] == nil and
+            IsString(resource.attributes["rum.sessionId"]) and
+            Len(resource.attributes["rum.sessionId"]) > 0
   resourcedetection:
     detectors:
       - env
@@ -183,10 +222,12 @@ service:
   # merge wins. See https://github.com/hyperdxio/hyperdx/pull/2351.
   pipelines:
     traces:
-      processors: [memory_limiter, batch]
+      processors: [memory_limiter, transform/session_id, batch]
     metrics:
       processors: [memory_limiter, batch]
+    # transform/session_id runs after transform so a `session.id` parsed from
+    # a JSON body wins. Replay rows skip transform but still need the copy.
     logs/out-default:
-      processors: [memory_limiter, transform, batch]
+      processors: [memory_limiter, transform, transform/session_id, batch]
     logs/out-rrweb:
-      processors: [memory_limiter, batch]
+      processors: [memory_limiter, transform/session_id, batch]

docker/otel-collector/custom.config.yaml

@@ -24,13 +24,13 @@ receivers:
 # service:
 #   pipelines:
 #     traces:
-#       processors: [memory_limiter/custom, batch]
+#       processors: [memory_limiter/custom, transform/session_id, batch]
 #     metrics:
 #       processors: [memory_limiter/custom, batch]
 #     logs/out-default:
-#       processors: [memory_limiter/custom, transform, batch]
+#       processors: [memory_limiter/custom, transform, transform/session_id, batch]
 #     logs/out-rrweb:
-#       processors: [memory_limiter/custom, batch]
+#       processors: [memory_limiter/custom, transform/session_id, batch]
 #
 # Example: tune `batch` for lower-latency exports.
 # (See packages/otel-collector/README.md for env-var alternatives.)
@@ -44,15 +44,15 @@ receivers:
 # service:
 #   pipelines:
 #     traces:
-#       processors: [memory_limiter, batch/lowlatency]
+#       processors: [memory_limiter, transform/session_id, batch/lowlatency]
 #     logs/out-default:
-#       processors: [memory_limiter, transform, batch/lowlatency]
+#       processors: [memory_limiter, transform, transform/session_id, batch/lowlatency]
 exporters:
   # fork data to otel-collector-json service (DEV ONLY)
   otlphttp/json:
     endpoint: http://otel-collector-json:4318
     headers:
-      authorization: "super-secure-ingestion-api-key"
+      authorization: 'super-secure-ingestion-api-key'
     compression: gzip
 service:
   pipelines:

docker/otel-collector/schema/seed/00002_otel_logs.sql

@@ -25,10 +25,12 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.otel_logs
   `__hdx_materialized_k8s.pod.name` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.name'] CODEC(ZSTD(3)),
   `__hdx_materialized_k8s.pod.uid` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.uid'] CODEC(ZSTD(3)),
   `__hdx_materialized_deployment.environment.name` LowCardinality(String) MATERIALIZED ResourceAttributes['deployment.environment.name'] CODEC(ZSTD(3)),
+  `__hdx_materialized_session.id` String MATERIALIZED LogAttributes['session.id'] CODEC(ZSTD(3)),
   `ResourceAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), ResourceAttributes::Array(Tuple(String, String))),
   `ScopeAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), ScopeAttributes::Array(Tuple(String, String))),
   `LogAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), LogAttributes::Array(Tuple(String, String))),
   INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array'),
+  INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array'),
   INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array'),
   INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array'),
   INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE text(tokenizer = 'array'),

docker/otel-collector/schema/seed/00002_otel_logs_compat.sql

@@ -26,7 +26,9 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.otel_logs
   `__hdx_materialized_k8s.pod.name` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.name'] CODEC(ZSTD(3)),
   `__hdx_materialized_k8s.pod.uid` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.uid'] CODEC(ZSTD(3)),
   `__hdx_materialized_deployment.environment.name` LowCardinality(String) MATERIALIZED ResourceAttributes['deployment.environment.name'] CODEC(ZSTD(3)),
+  `__hdx_materialized_session.id` String MATERIALIZED LogAttributes['session.id'] CODEC(ZSTD(3)),
   INDEX idx_trace_id TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_session_id __hdx_materialized_session.id TYPE bloom_filter(0.001) GRANULARITY 1,
   INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
   INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
   INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,

docker/otel-collector/schema/seed/00004_hyperdx_sessions.sql

@@ -17,7 +17,11 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.hyperdx_sessions
   `ScopeVersion` LowCardinality(String) CODEC(ZSTD(3)),
   `ScopeAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(3)),
   `LogAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(3)),
+  `__hdx_materialized_session.id` String MATERIALIZED LogAttributes['session.id'] CODEC(ZSTD(3)),
+  `__hdx_materialized_rum.sessionId` String MATERIALIZED ResourceAttributes['rum.sessionId'] CODEC(ZSTD(3)),
   INDEX idx_trace_id TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array'),
+  INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE text(tokenizer = 'array'),
   INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
   INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
   INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,

docker/otel-collector/schema/seed/00004_hyperdx_sessions_compat.sql

@@ -0,0 +1,40 @@
+-- +goose Up
+-- Compatibility schema for ClickHouse < 26.2 (no full text search indexes)
+CREATE TABLE IF NOT EXISTS ${DATABASE}.hyperdx_sessions
+(
+  `Timestamp` DateTime64(9) CODEC(Delta, ZSTD(3)),
+  `TimestampTime` DateTime DEFAULT toDateTime(Timestamp),
+  `TraceId` String CODEC(ZSTD(3)),
+  `SpanId` String CODEC(ZSTD(3)),
+  `TraceFlags` UInt8,
+  `SeverityText` LowCardinality(String) CODEC(ZSTD(3)),
+  `SeverityNumber` UInt8,
+  `ServiceName` LowCardinality(String) CODEC(ZSTD(3)),
+  `Body` String CODEC(ZSTD(3)),
+  `ResourceSchemaUrl` LowCardinality(String) CODEC(ZSTD(3)),
+  `ResourceAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(3)),
+  `ScopeSchemaUrl` LowCardinality(String) CODEC(ZSTD(3)),
+  `ScopeName` String CODEC(ZSTD(3)),
+  `ScopeVersion` LowCardinality(String) CODEC(ZSTD(3)),
+  `ScopeAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(3)),
+  `LogAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(3)),
+  `__hdx_materialized_session.id` String MATERIALIZED LogAttributes['session.id'] CODEC(ZSTD(3)),
+  `__hdx_materialized_rum.sessionId` String MATERIALIZED ResourceAttributes['rum.sessionId'] CODEC(ZSTD(3)),
+  INDEX idx_trace_id TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_session_id __hdx_materialized_session.id TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_scope_attr_value mapValues(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_log_attr_value mapValues(LogAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
+  INDEX idx_lower_body lower(Body) TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 8
+)
+ENGINE = MergeTree
+PARTITION BY toDate(TimestampTime)
+PRIMARY KEY (ServiceName, TimestampTime)
+ORDER BY (ServiceName, TimestampTime, Timestamp)
+TTL TimestampTime + ${SESSIONS_TTL}
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
+

packages/otel-collector/README.md

@@ -433,13 +433,13 @@ processors:
 service:
   pipelines:
     traces:
-      processors: [memory_limiter/custom, batch]
+      processors: [memory_limiter/custom, transform/session_id, batch]
     metrics:
       processors: [memory_limiter/custom, batch]
     logs/out-default:
-      processors: [memory_limiter/custom, transform, batch]
+      processors: [memory_limiter/custom, transform, transform/session_id, batch]
     logs/out-rrweb:
-      processors: [memory_limiter/custom, batch]
+      processors: [memory_limiter/custom, transform/session_id, batch]
 ```
 
 After restart, the collector instantiates `memory_limiter/custom` (and not
@@ -466,9 +466,9 @@ processors:
 service:
   pipelines:
     traces:
-      processors: [memory_limiter, batch/lowlatency]
+      processors: [memory_limiter, transform/session_id, batch/lowlatency]
     logs/out-default:
-      processors: [memory_limiter, transform, batch/lowlatency]
+      processors: [memory_limiter, transform, transform/session_id, batch/lowlatency]
 ```
 
 Notes:

packages/otel-collector/cmd/migrate/main.go

@@ -718,6 +718,37 @@ func removeCompatTracesSchema(tempDir string) error {
 	return nil
 }
 
+// swapSessionsSchemaForCompat replaces the full-text-search sessions schema
+// with the compatibility variant (bloom_filter session id indexes) in the
+// processed temp directory. It removes 00004_hyperdx_sessions.sql and renames
+// 00004_hyperdx_sessions_compat.sql to take its place, so goose runs the
+// compat schema instead.
+func swapSessionsSchemaForCompat(tempDir string) error {
+	fullTextPath := filepath.Join(tempDir, "00004_hyperdx_sessions.sql")
+	compatPath := filepath.Join(tempDir, "00004_hyperdx_sessions_compat.sql")
+
+	if err := os.Remove(fullTextPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to remove full text sessions schema: %w", err)
+	}
+
+	if err := os.Rename(compatPath, fullTextPath); err != nil {
+		return fmt.Errorf("failed to rename compat sessions schema: %w", err)
+	}
+
+	return nil
+}
+
+// removeCompatSessionsSchema removes the compat sessions schema file from the
+// temp directory when full text search is supported, so goose doesn't run both
+// schemas.
+func removeCompatSessionsSchema(tempDir string) error {
+	compatPath := filepath.Join(tempDir, "00004_hyperdx_sessions_compat.sql")
+	if err := os.Remove(compatPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to remove compat sessions schema: %w", err)
+	}
+	return nil
+}
+
 // listSQLFiles lists all SQL files in a directory for logging purposes
 func listSQLFiles(dir string) ([]string, error) {
 	entries, err := os.ReadDir(dir)
@@ -796,22 +827,28 @@ func main() {
 	}
 	defer os.RemoveAll(tempDir)
 
-	// Select the appropriate logs and traces schemas based on ClickHouse version
+	// Select the appropriate logs, traces and sessions schemas based on ClickHouse version
 	if supportsFullTextSearch(chMajor, chMinor) {
 		if err := removeCompatLogsSchema(tempDir); err != nil {
 			log.Fatalf("Failed to remove compat logs schema: %v", err)
 		}
 		if err := removeCompatTracesSchema(tempDir); err != nil {
 			log.Fatalf("Failed to remove compat traces schema: %v", err)
 		}
+		if err := removeCompatSessionsSchema(tempDir); err != nil {
+			log.Fatalf("Failed to remove compat sessions schema: %v", err)
+		}
 	} else {
-		log.Printf("ClickHouse %d.%d < 26.2, falling back to compatibility logs and traces schemas", chMajor, chMinor)
+		log.Printf("ClickHouse %d.%d < 26.2, falling back to compatibility logs, traces and sessions schemas", chMajor, chMinor)
 		if err := swapLogsSchemaForCompat(tempDir); err != nil {
 			log.Fatalf("Failed to swap logs schema: %v", err)
 		}
 		if err := swapTracesSchemaForCompat(tempDir); err != nil {
 			log.Fatalf("Failed to swap traces schema: %v", err)
 		}
+		if err := swapSessionsSchemaForCompat(tempDir); err != nil {
+			log.Fatalf("Failed to swap sessions schema: %v", err)
+		}
 	}
 
 	if os.Getenv("ENABLE_PROMQL") != "true" {

packages/otel-collector/cmd/migrate/main_test.go

@@ -1133,6 +1133,81 @@ func TestRemoveCompatLogsSchema(t *testing.T) {
 	})
 }
 
+func TestSwapSessionsSchemaForCompat(t *testing.T) {
+	t.Run("swaps compat over full text", func(t *testing.T) {
+		dir := t.TempDir()
+		fullTextPath := filepath.Join(dir, "00004_hyperdx_sessions.sql")
+		compatPath := filepath.Join(dir, "00004_hyperdx_sessions_compat.sql")
+
+		if err := os.WriteFile(fullTextPath, []byte("FULL TEXT SCHEMA"), 0644); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(compatPath, []byte("COMPAT SCHEMA"), 0644); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := swapSessionsSchemaForCompat(dir); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		got, err := os.ReadFile(fullTextPath)
+		if err != nil {
+			t.Fatalf("failed to read swapped file: %v", err)
+		}
+		if string(got) != "COMPAT SCHEMA" {
+			t.Errorf("expected compat content, got %q", string(got))
+		}
+		if _, err := os.Stat(compatPath); !os.IsNotExist(err) {
+			t.Error("compat file should have been renamed away")
+		}
+	})
+
+	t.Run("errors when compat file is missing", func(t *testing.T) {
+		dir := t.TempDir()
+		fullTextPath := filepath.Join(dir, "00004_hyperdx_sessions.sql")
+
+		if err := os.WriteFile(fullTextPath, []byte("FULL TEXT SCHEMA"), 0644); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := swapSessionsSchemaForCompat(dir); err == nil {
+			t.Fatal("expected error when compat file is missing")
+		}
+	})
+}
+
+func TestRemoveCompatSessionsSchema(t *testing.T) {
+	t.Run("removes compat file and preserves others", func(t *testing.T) {
+		dir := t.TempDir()
+		fullTextPath := filepath.Join(dir, "00004_hyperdx_sessions.sql")
+		compatPath := filepath.Join(dir, "00004_hyperdx_sessions_compat.sql")
+
+		if err := os.WriteFile(fullTextPath, []byte("FULL TEXT"), 0644); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(compatPath, []byte("COMPAT"), 0644); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := removeCompatSessionsSchema(dir); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		if _, err := os.Stat(compatPath); !os.IsNotExist(err) {
+			t.Error("compat file should have been removed")
+		}
+		if _, err := os.Stat(fullTextPath); err != nil {
+			t.Error("full text file should still exist")
+		}
+	})
+
+	t.Run("no error when compat file does not exist", func(t *testing.T) {
+		if err := removeCompatSessionsSchema(t.TempDir()); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
 // ---------------------------------------------------------------------------
 // processSchemaDir
 // ---------------------------------------------------------------------------