Merge pull request #587 from metrico/fix/585

fix/585

Author: akvlad

.gitignore

@@ -9,3 +9,5 @@ node_modules
 /wasm_parts/vendor/
 /wasm_parts/main.wasm
 /wasm_parts/wasm_parts.iml
+/test/qryn_test_env/clickhouse/_data/
+/test/qryn_test_env/grafana/_data/

parser/registry/smart_optimizations/optimization_v3_2.js

@@ -1,4 +1,4 @@
-const { getDuration, preJoinLabels, dist } = require('../common')
+const { getDuration, preJoinLabels, dist, sharedParamNames } = require('../common')
 const reg = require('./log_range_agg_reg_v3_2')
 const Sql = require('@cloki/clickhouse-sql')
 const { DATABASE_NAME, checkVersion } = require('../../../lib/utils')
@@ -44,24 +44,30 @@ function isLogPipeline (token) {
  */
 module.exports.apply = (token, fromNS, toNS, stepNS) => {
   fromNS = Math.floor(fromNS / 15000000000) * 15000000000
+  const fromParam = new Sql.Parameter(sharedParamNames.from)
+  const toParam = new Sql.Parameter(sharedParamNames.to)
   const tsClause = toNS
     ? Sql.between('samples.timestamp_ns', fromNS, toNS)
     : Sql.Gt('samples.timestamp_ns', fromNS)
   let q = (new Sql.Select())
     .select(['samples.fingerprint', 'fingerprint'])
     .from([`${DATABASE_NAME()}.metrics_15s${_dist}`, 'samples'])
     .where(tsClause)
+    .addParam(fromParam)
+    .addParam(toParam)
+  fromParam.set(fromNS)
+  toParam.set(toNS)
 
   q.ctx = {
     step: stepNS / 1000000000,
     inline: !!clusterName
   }
 
-  preJoinLabels(token, q, dist)
-
   for (const streamSelectorRule of token.Children('log_stream_selector_rule')) {
     q = streamSelectorReg[streamSelectorRule.Child('operator').value](streamSelectorRule, q)
   }
+  preJoinLabels(token, q, dist)
+  q = q.groupBy('labels')
 
   const lra = token.Child('log_range_aggregation')
   q = reg[lra.Child('log_range_aggregation_fn').value](lra, q)

promql/index.js

@@ -61,7 +61,9 @@ module.exports.series = async (query, fromMs, toMs) => {
     const data = await rawRequest(req.toString() + ' FORMAT JSON',
       null,
       DATABASE_NAME())
-    return data.data.data.map(l => JSON.parse(l.labels))
+    return data.data.data.map(l =>
+      Object.fromEntries(Object.entries(JSON.parse(l.labels)).filter(e => e[1]))
+    )
   } catch (e) {
     if (e instanceof prometheus.WasmError) {
       throw new PSQLError(e.message)

test/e2e

@@ -0,0 +1,121 @@
+version: '2.1'
+
+networks:
+  qryn-test:
+    driver: bridge
+
+services:
+  grafana:
+    image: grafana/grafana:11.2.2
+    container_name: grafana
+    user: root
+    networks:
+      - qryn-test
+    volumes:
+      - ./grafana/_data:/var/lib/grafana:rw
+      - ./grafana/provisioning/:/etc/grafana/provisioning/
+    environment:
+      - GF_SECURITY_ADMIN_USER=${ADMIN_USER:-admin}
+      - GF_SECURITY_ADMIN_PASSWORD=${ADMIN_PASSWORD:-admin}
+      - GF_USERS_ALLOW_SIGN_UP=false
+      - GF_USERS_DEFAULT_THEME=light
+      - GF_EXPLORE_ENABLED=true
+      - GF_ALERTING_ENABLED=false
+      - GF_UNIFIED_ALERTING_ENABLED=true
+      - GF_FEATURE_TOGGLES_ENABLE=traceToMetrics,publicDashboards,tempoApmTable
+      - GF_INSTALL_PLUGINS=grafana-pyroscope-app,https://storage.googleapis.com/integration-artifacts/grafana-lokiexplore-app/grafana-lokiexplore-app-latest.zip;grafana-lokiexplore-app
+    restart: unless-stopped
+    ports:
+      - 3000:3000
+    depends_on:
+      - qryn
+
+  clickhouse-server:
+    image: clickhouse/clickhouse-server:24.1
+    container_name: clickhouse-server
+    hostname: clickhouse
+    restart: unless-stopped
+    volumes:
+      - ./clickhouse/_data:/var/lib/clickhouse
+    networks:
+      - qryn-test
+    environment:
+      - CLICKHOUSE_USER=qryn
+      - CLICKHOUSE_PASSWORD=demo
+    ports:
+      - 8123:8123
+      - 9000:9000
+    healthcheck:
+      test: ['CMD', 'wget', '--spider', '-q', '127.0.0.1:8123/ping']
+      interval: 1s
+      timeout: 1s
+      retries: 30
+
+  qryn:
+    image: node:22
+    container_name: qryn
+    hostname: qryn
+    restart: unless-stopped
+    volumes:
+      - ../../:/app
+    networks:
+      - qryn-test
+    expose:
+      - 3100
+    ports:
+      - "3100:3100"
+    environment:
+      - CLICKHOUSE_SERVER=clickhouse-server
+      - CLICKHOUSE_PORT=8123
+      - CLICKHOUSE_AUTH=qryn:demo
+      - CLICKHOUSE_DB=qryn
+      - NODE_OPTIONS="--max-old-space-size=4096"
+      - FASTIFY_METRICS=true
+    working_dir: /app
+    entrypoint: sh
+    command:
+      - -c
+      - "mkdir /_app && cp -rf patches package.json package-lock.json /_app && cd /_app && npm install && cd /app && NODE_PATH='.:../_app/node_modules' node qryn_node.js"
+    depends_on:
+      clickhouse-server:
+        condition: service_healthy
+
+  longtest:
+    image: golang:1.22-alpine
+    volumes:
+      - ./longtest:/longtest
+    working_dir: /longtest
+    environment:
+      MODE: LMZ
+      URL: http://qryn:3100
+    command: ["go", "run", "."]
+    networks:
+      - qryn-test
+    depends_on:
+      - qryn
+
+  otel-collector:
+    container_name: otel-collector
+    hostname: otel-collector
+    image: ghcr.io/metrico/qryn-otel-collector:0.0.5
+    networks:
+      - qryn-test
+    volumes:
+      - ./otel/otel-collector-config.yaml:/etc/otel/config.yaml
+    ports:
+      - "3200:3100"     # Loki/Logql HTTP receiver
+      - "3201:3200"     # Loki/Logql gRPC receiver
+      - "8088:8088"     # Splunk HEC receiver
+      - "5514:5514"     # Syslog TCP Rereceiverceiver
+      - "24224:24224"   # Fluent Forward receiver
+      - "4317:4317"     # OTLP gRPC receiver
+      - "4318:4318"     # OTLP HTTP receiver
+      - "14250:14250"   # Jaeger gRPC receiver
+      - "14268:14268"   # Jaeger thrift HTTP receiver
+      - "9411:9411"     # Zipkin Trace receiver
+      - "11800:11800"   # Skywalking gRPC receiver
+      - "12800:12800"   # Skywalking HTTP receiver
+      - "8086:8086"     # InfluxDB Line proto HTTP
+      - "8062:8062"     # Pyroscope jprof
+    restart: on-failure
+