SqlObfuscator fix aggregator issues, log aggregator and other services in dev mode

.changeset/odd-geckos-worry.md

@@ -0,0 +1,5 @@
+---
+'@hyperdx/api': patch
+---
+
+Add experimental sql obfuscator

docker-compose.dev.yml

@@ -75,6 +75,7 @@ services:
  context: ./docker/otel-collector
  target: dev
  environment:
+ HYPERDX_API_KEY: ${HYPERDX_API_KEY}
  HYPERDX_LOG_LEVEL: ${HYPERDX_LOG_LEVEL}
  volumes:
  - ./docker/otel-collector/config.yaml:/etc/otelcol-contrib/config.yaml
@@ -99,12 +100,25 @@ services:
  ports:
  - 8001:8001
  environment:
+ CLICKHOUSE_LOG_LEVEL: ${HYPERDX_LOG_LEVEL}
+ EXPRESS_SESSION_SECRET: 'hyperdx is cool 👋'
+ HDX_NODE_ADVANCED_NETWORK_CAPTURE: 1
+ HDX_NODE_BETA_MODE: 1
+ HDX_NODE_CONSOLE_CAPTURE: 1
+ SENTRY_DSN: ${SENTRY_DSN}
+ INGESTOR_API_URL: 'http://ingestor:8002'
+ MINER_API_URL: 'http://miner:5123'
+ OTEL_EXPORTER_OTLP_ENDPOINT: 'http://otel-collector:4318'
+ OTEL_SERVICE_NAME: 'hdx-oss-dev-aggregator'
  APP_TYPE: 'aggregator'
  CLICKHOUSE_HOST: http://ch-server:8123
  CLICKHOUSE_PASSWORD: aggregator
  CLICKHOUSE_USER: aggregator
  FRONTEND_URL: 'http://localhost:8080' # need to be localhost (CORS)
  HYPERDX_LOG_LEVEL: ${HYPERDX_LOG_LEVEL}
+ # uncomment to allow aggregator logs
+ # disabled by default due to possible infinite loop (aggregator aggregating aggregator logs)
+ # HYPERDX_API_KEY: ${HYPERDX_API_KEY}
  MONGO_URI: 'mongodb://db:27017/hyperdx'
  NODE_ENV: development
  PORT: 8001

packages/api/src/config.ts

@@ -26,3 +26,4 @@ export const PORT = Number.parseInt(env.PORT as string);
 export const REDIS_URL = env.REDIS_URL as string;
 export const SERVER_URL = env.SERVER_URL as string;
 export const USAGE_STATS_ENABLED = env.USAGE_STATS_ENABLED !== 'false';
+export const OBFUSCATE_SQL = env.OBFUSCATE_SQL !== 'false';

packages/api/src/gobin/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Datadog, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

packages/api/src/gobin/src/go.mod

@@ -0,0 +1,5 @@
+module sqlObfuscator
+
+go 1.21.5
+
+require github.com/DataDog/go-sqllexer v0.0.10

packages/api/src/gobin/src/go.sum

@@ -0,0 +1,10 @@
+github.com/DataDog/go-sqllexer v0.0.10 h1:u07DuRfdlPPmOX/dclb1gcn/zaqWxUiURRRVenKILxc=
+github.com/DataDog/go-sqllexer v0.0.10/go.mod h1:KwkYhpFEVIq+BfobkTC1vfqm4gTi65skV/DpDBXtexc=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

packages/api/src/gobin/src/sql_obfuscator.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/DataDog/go-sqllexer"
+)
+
+// This is a simple command line tool that reads multiple newline-separated SQL queries from stdin
+// and normalizes and obfuscates them, then prints them one at a time (newline separated) to stdout.
+// Example:
+// $ echo "SELECT * FROM foo as foo_table limit 1; SELECT * FROM /* sql comment */ bar where name = 'bob';" | go run sql_obfuscator.go
+// SELECT * FROM foo limit ?; SELECT * FROM bar where name = ?
+func main() {
+	reader := bufio.NewReader(os.Stdin)
+	for {
+ query, err := reader.ReadString('\n')
+ if err != nil {
+ // write to stderr
+ fmt.Fprintf(os.Stderr, "error: %s\n", err)
+ os.Exit(1)
+ }
+ stripped_string := strings.Trim(query, " \t\r\n")
+ if stripped_string == "" {
+ // skip empty entries
+ continue
+ }
+
+ normalizer := sqllexer.NewNormalizer(
+ sqllexer.WithCollectComments(false),
+ sqllexer.WithCollectCommands(false),
+ sqllexer.WithCollectTables(false),
+ sqllexer.WithKeepSQLAlias(false),
+ )
+
+ normalized, _, err := normalizer.Normalize(query)
+ if err != nil {
+ // write to stderr
+ fmt.Fprintf(os.Stderr, "error: %s\n", err)
+ os.Exit(1)
+ }
+
+ obfuscator := sqllexer.NewObfuscator()
+ obfuscated := obfuscator.Obfuscate(normalized)
+ fmt.Printf("%s\n", obfuscated)
+	}
+}

packages/api/src/routers/aggregator/root.ts

@@ -29,7 +29,7 @@ const bulkInsert = async (
  switch (hdxTelemetry) {
  case 'metric':
  await bulkInsertTeamMetricStream(
- vectorMetricParser.parse(data as VectorMetric[]),
+await vectorMetricParser.parse(data as VectorMetric[]),
  );
  break;
  default: {
@@ -46,12 +46,12 @@ const bulkInsert = async (
  bulkInsertTeamLogStream(
  team.logStreamTableVersion,
  team._id.toString(),
- vectorLogParser.parse(logs),
+await vectorLogParser.parse(logs),
  ),
  ];
  if (rrwebEvents.length > 0) {
  promises.push(
- bulkInsertRrwebEvents(vectorRrwebParser.parse(rrwebEvents)),
+ bulkInsertRrwebEvents(await vectorRrwebParser.parse(rrwebEvents)),
  );
  }
  await Promise.all(promises);

packages/api/src/utils/__tests__/__snapshots__/logParser.test.ts.snap

@@ -1,6 +1,6 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`logParser mapObjectToKeyValuePairs 1`] = `
+exports[`logParser mapObjectToKeyValuePairs obeys basic serialization 1`] = `
 Object {
  "bool.names": Array [
  "foo2",
@@ -31,7 +31,7 @@ Object {
 }
 `;
 
-exports[`logParser mapObjectToKeyValuePairs 2`] = `
+exports[`logParser mapObjectToKeyValuePairs obeys basic serialization 2`] = `
 Object {
  "bool.names": Array [],
  "bool.values": Array [],

packages/api/src/utils/__tests__/logParser.test.ts

@@ -53,69 +53,90 @@ describe('logParser', () => {
  ]);
  });
 
- it('mapObjectToKeyValuePairs', () => {
- expect(mapObjectToKeyValuePairs(null as any)).toEqual({
- 'bool.names': [],
- 'bool.values': [],
- 'number.names': [],
- 'number.values': [],
- 'string.names': [],
- 'string.values': [],
- });
+ describe('mapObjectToKeyValuePairs', () => {
+ it('obeys basic serialization', async () => {
+ expect(await mapObjectToKeyValuePairs(null as any)).toEqual({
+ 'bool.names': [],
+ 'bool.values': [],
+ 'number.names': [],
+ 'number.values': [],
+ 'string.names': [],
+ 'string.values': [],
+ });
 
- expect(mapObjectToKeyValuePairs({})).toEqual({
- 'bool.names': [],
- 'bool.values': [],
- 'number.names': [],
- 'number.values': [],
- 'string.names': [],
- 'string.values': [],
- });
+expect(await mapObjectToKeyValuePairs({})).toEqual({
+'bool.names': [],
+'bool.values': [],
+'number.names': [],
+'number.values': [],
+'string.names': [],
+'string.values': [],
+});
 
- expect(
- mapObjectToKeyValuePairs({ foo: '123', foo1: 123, foo2: false }),
- ).toEqual({
- 'bool.names': ['foo2'],
- 'bool.values': [0],
- 'number.names': ['foo1'],
- 'number.values': [123],
- 'string.names': ['foo'],
- 'string.values': ['123'],
- });
+expect(
+ await mapObjectToKeyValuePairs({ foo: '123', foo1: 123, foo2: false }),
+).toEqual({
+'bool.names': ['foo2'],
+'bool.values': [0],
+'number.names': ['foo1'],
+'number.values': [123],
+'string.names': ['foo'],
+'string.values': ['123'],
+});
 
- expect(
- mapObjectToKeyValuePairs({
- foo: '123',
- foo1: 123,
- foo2: false,
- nested: { foo: 'bar' },
- good: {
- burrito: {
- is: true,
- },
- },
- array1: [456],
- array2: [
- 'foo1',
- {
- foo2: 'bar2',
+ expect(
+ await mapObjectToKeyValuePairs({
+ foo: '123',
+ foo1: 123,
+ foo2: false,
+ nested: { foo: 'bar' },
+ good: {
+ burrito: {
+ is: true,
+ },
  },
- [
+ array1: [456],
+ array2: [
+ 'foo1',
  {
-foo3: 'bar3',
+foo2: 'bar2',
  },
+ [
+ {
+ foo3: 'bar3',
+ },
+ ],
  ],
- ],
- }),
- ).toMatchSnapshot();
+ }),
+ ).toMatchSnapshot();
 
- const testObject = {};
- for (let i = 0; i < 2000; i++) {
- testObject[`foo${i}`] = i;
- }
- const result = mapObjectToKeyValuePairs(testObject);
- expect(result['number.names'].length).toEqual(1024);
- expect(result['number.values'].length).toEqual(1024);
- expect(result).toMatchSnapshot();
+ const testObject = {};
+ for (let i = 0; i < 2000; i++) {
+ testObject[`foo${i}`] = i;
+ }
+ const result = await mapObjectToKeyValuePairs(testObject);
+ expect(result['number.names'].length).toEqual(1024);
+ expect(result['number.values'].length).toEqual(1024);
+ expect(result).toMatchSnapshot();
+ });
+
+ it('adds obfuscated sql to output', async () => {
+ const original = {
+ db: {
+ statement: `SELECT * from users where name = 'Carl' limit 1 order by name asc;\n`,
+ },
+ };
+ expect(await mapObjectToKeyValuePairs(original)).toEqual({
+ 'bool.names': [],
+ 'bool.values': [],
+ 'number.names': [],
+ 'number.values': [],
+ 'string.names': ['db.statement', 'db.sql.normalized'],
+ 'string.values': [
+ original.db.statement,
+ `SELECT * from users where name = ? limit ? order by name asc\n`,
+ ],
+ });
+ });
  });
 });

packages/api/src/utils/__tests__/sqlObfuscator.test.ts

@@ -0,0 +1,53 @@
+import * as childProcess from 'child_process';
+
+import { getChild, sqlObfuscator } from '../sqlObfuscator';
+
+describe('logParser', () => {
+ it('obfuscates a basic query', async () => {
+ const n = 1;
+ const start = Date.now();
+ for (let i = 0; i < n; i++) {
+ expect(
+ await sqlObfuscator(
+ `SELECT * from users where name = 'Carl' limit 1 order by name asc;\n`,
+ ),
+ ).toEqual(
+ `SELECT * from users where name = ? limit ? order by name asc`.replace(
+ /(\r\n|\n|\r)/gm,
+ ' ',
+ ) + '\n',
+ );
+ }
+ const end = Date.now();
+ //console.log(`Took ${(end - start) / n}ms`);
+ });
+
+ it('handles multiple statements in one line', async () => {
+ expect(
+ await sqlObfuscator(
+ `SELECT * from users as should_be_removed where name = 'Carl' limit 1 order by name asc; SELECT * from users /* standard comment, really */ where name = 'Carl' limit 1 order by name asc;\n`,
+ ),
+ ).toEqual(
+ `SELECT * from users where name = ? limit ? order by name asc; SELECT * from users where name = ? limit ? order by name asc`.replace(
+ /(\r\n|\n|\r)/gm,
+ ' ',
+ ) + '\n',
+ );
+ });
+
+ it('handles if the subprocess dies', async () => {
+ const subprocess = getChild();
+ subprocess.kill();
+ expect(
+ await sqlObfuscator(
+ `SELECT * from users where name = 'Carl' limit 1 order by name asc;\n`,
+ ),
+ ).toEqual(
+ `SELECT * from users where name = ? limit ? order by name asc`.replace(
+ /(\r\n|\n|\r)/gm,
+ ' ',
+ ) + '\n',
+ );
+ expect(subprocess).not.toBe(getChild());
+ });
+});