Enrich process telemetry with binary memory metrics and fix stack_id in OTEL resource

.changeset/eleven-nails-argue.md

@@ -0,0 +1,5 @@
+---
+'@core/electric-telemetry': patch
+---
+
+Add binary memory, average number of off-heap binaries and their ref counts to top processes by memory metric.

.changeset/forty-pillows-laugh.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Include stack_id in otel opts for stack metrics. It had been omitted by mistake before.

.github/workflows/integration_tests.yml

@@ -28,6 +28,7 @@ jobs:
       - uses: actions/checkout@v4
 
       - uses: erlef/setup-beam@v1
+        id: setup_beam
         with:
           version-type: strict
           version-file: '.tool-versions'
@@ -62,6 +63,15 @@ jobs:
         run: mix compile
         working-directory: packages/sync-service
 
+      - name: Cache lux
+        uses: actions/cache@v4
+        with:
+          path: integration-tests/lux
+          key: '${{ runner.os }}-lux-${{ steps.setup_beam.outputs.otp-version }}'
+          restore-keys: |
+            ${{ runner.os }}-lux-${{ steps.setup_beam.outputs.otp-version }}
+            ${{ runner.os }}-lux
+
       - name: Setup lux
         run: make
 

integration-tests/tests/_macros.luxinc

@@ -231,11 +231,16 @@
       -p 4318:4318 \
       -v $(realpath ../support_files/otel-collector-config.yaml):/conf/otel-collector-config.yaml \
       otel/opentelemetry-collector-contrib --config=/conf/otel-collector-config.yaml
+
     # Allow time for docker to pull the image if not cached
     [timeout 120]
+
     ??Starting HTTP server
     ?"endpoint": "(0\.0\.0\.0|\[::\]):4318"
     ??Everything is ready. Begin running and processing data.
+
+    # Reset the timeout for subsequent pattern matching
+    [timeout 10]
 [endmacro]
 
 [macro teardown_container container_name]

integration-tests/tests/otel-export.lux

@@ -0,0 +1,232 @@
+[doc Verify that application and stack metrics are correctly exported via Otel]
+
+[include _macros.luxinc]
+
+[global pg_container_name=otel-export__pg]
+
+###
+
+[invoke setup_otel_collector]
+
+[invoke setup_pg]
+
+[invoke setup_electric_with_env "ELECTRIC_OTLP_ENDPOINT=http://localhost:4318 ELECTRIC_SYSTEM_METRICS_POLL_INTERVAL=1s ELECTRIC_STACK_TELEMETRY_INIT_DELAY=1s ELECTRIC_OTEL_EXPORT_PERIOD=2s DO_NOT_START_CONN_MAN_PING=1 ELECTRIC_LOG_LEVEL=info OTEL_RESOURCE_ATTRIBUTES=custom.attr=electric.val"]
+
+# Spawn a process containing off-heap binary references and ensure its in the top 5 by memory footprint.
+# Otel Collector sorts metrics in its debug output, so the process label for our process needs
+# to be first lexicographically for easier output matching further down.
+[shell electric]
+  """!
+  _pid = spawn_link(fn ->
+    Process.set_label(:A_memory_hog)
+
+    on_heap_strings =
+      Enum.map(1..4000, fn _ -> String.duplicate("on heap", 4000) end)
+
+    off_heap_strings =
+      Enum.map(1..10, fn i -> String.duplicate("1234567890", 7000) end)
+
+    receive do
+      pid -> send(pid, {on_heap_strings, off_heap_strings})
+    end
+  end)
+  """
+
+  ??#PID
+
+[shell otel_collector]
+  # Verify that the Collector receives expected application metrics from Electric
+  """?
+  info	ResourceMetrics #0
+  Resource SchemaURL: 
+  Resource attributes:
+       -> custom.attr: Str\(electric.val\)
+       -> instance.id: Str\([-a-f0-9]+\)
+       -> name: Str\(metrics\)
+       -> service.name: Str\(electric\)
+       -> service.version: Str\([0-9.]+\)
+  ScopeMetrics #0
+  """
+
+  # Verify the presence of process.bin_memory.* metrics
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.bin_memory\.avg_bin_count
+       -> Description: 
+       -> Unit: 
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [.0-9]+
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.bin_memory\.avg_ref_count
+       -> Description: 
+       -> Unit: 
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: 1\.000000
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.bin_memory\.max_bin_count
+       -> Description: 
+       -> Unit: 
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [0-9]+
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.bin_memory\.max_ref_count
+       -> Description: 
+       -> Unit: 
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [0-9]+
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.bin_memory\.total
+       -> Description: 
+       -> Unit: By
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [0-9]+
+  """
+
+  # Verify the presence of process.memory.total metric
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: process\.memory\.total
+       -> Description: 
+       -> Unit: By
+       -> DataType: Gauge
+  NumberDataPoints #0
+  Data point attributes:
+       -> process_type: Str\(A_memory_hog\)
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [0-9]+
+  """
+
+  # Verify that the Collector receives stack metrics from Electric with expected resource attributes
+  """?
+  info	ResourceMetrics #0
+  Resource SchemaURL: 
+  Resource attributes:
+       -> custom.attr: Str\(electric.val\)
+       -> instance.id: Str\([-a-f0-9]+\)
+       -> name: Str\(metrics\)
+       -> service.name: Str\(electric\)
+       -> service.version: Str\([0-9.]+\)
+       -> stack_id: Str\(single_stack\)
+  ScopeMetrics #0
+  """
+
+  # Verify that LSN metrics are exported
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: electric\.postgres\.replication\.pg_wal_offset
+       -> Description: 
+       -> Unit: 
+       -> DataType: Gauge
+  NumberDataPoints #0
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [-+.0-9]+
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: electric\.postgres\.replication\.slot_confirmed_flush_lsn_lag
+       -> Description: 
+       -> Unit: By
+       -> DataType: Gauge
+  NumberDataPoints #0
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [-+.0-9]+
+  """
+
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: electric\.postgres\.replication\.slot_retained_wal_size
+       -> Description: 
+       -> Unit: By
+       -> DataType: Gauge
+  NumberDataPoints #0
+  StartTimestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Timestamp: [-0-9]+ [.:0-9]+ [-+0-9]+ UTC
+  Value: [-+.0-9]+
+  """
+
+[invoke start_psql]
+
+[shell psql]
+  !create table items (val text);
+  ??CREATE
+
+[shell client]
+  [invoke curl_shape "http://localhost:3000/v1/shape?table=items&offset=-1"]
+
+  ??HTTP/1.1 200 OK
+  ?electric-handle: ([\d-]+)
+  [global handle=$1]
+  ?electric-offset: ([\w\d_]+)
+  [global offset=$1]
+
+[shell psql]
+  !insert into items values ('3');
+  ??INSERT
+
+[shell client]
+  [invoke curl_shape "http://localhost:3000/v1/shape?table=items&handle=$handle&offset=$offset&live"]
+
+  ??HTTP/1.1 200 OK
+  ??"value":{"val":"3"}
+
+[shell otel_collector]
+  """?
+  Metric #[0-9]+
+  Descriptor:
+       -> Name: electric\.storage\.transaction_stored\.bytes
+  """
+
+###
+
+[cleanup]
+  [invoke teardown]