Upload parquet logs to S3

Author: platypii

package.json

@@ -19,9 +19,13 @@
     "eslint": "9.39.2",
     "eslint-plugin-jsdoc": "62.9.0",
     "globals": "16.2.0",
+    "hyparquet": "1.25.6",
     "typescript": "6.0.2",
     "vitest": "2.1.0"
   },
+  "optionalDependencies": {
+    "hyparquet-writer": "0.14.0"
+  },
   "keywords": [
     "opentelemetry",
     "otlp",

src/collector.js

@@ -48,34 +48,53 @@ const OTLP_NS_PER_MS = 1000000n
 const MIN_DATE_MS = -8640000000000000n
 const MAX_DATE_MS = 8640000000000000n
 
+/**
+ * @import { UploadOptions } from './upload/upload.d.ts'
+ */
+
 class Collector {
-  /** @param {{ port?: number, outputDir?: string }} [options] */
+  /** @param {{ port?: number, outputDir?: string, upload?: UploadOptions }} [options] */
   constructor(options = {}) {
     this.port = options.port ?? 4318
     this.outputDir = options.outputDir || './otel-data'
+    this.uploadOptions = options.upload
     /** @type {import('node:http').Server | null} */
     this.server = null
+    /** @type {{ start: () => Promise<void>, stop: () => Promise<void> } | null} */
+    this.uploader = null
   }
 
-  start() {
+  async start() {
     ensureDir(this.outputDir)
 
     const server = createServer(this.handleData.bind(this))
     this.server = server
 
-    return new Promise((resolve) => {
+    await new Promise((resolve) => {
       server.listen(this.port, () => resolve(undefined))
     })
+
+    if (this.uploadOptions) {
+      const { createUploader } = await import('./upload/index.js')
+      this.uploader = createUploader({
+        outputDir: this.outputDir,
+        options: this.uploadOptions,
+      })
+      await this.uploader.start()
+    }
   }
 
-  stop() {
-    return new Promise((resolve, reject) => {
+  async stop() {
+    if (this.uploader) {
+      await this.uploader.stop()
+      this.uploader = null
+    }
+    await new Promise((resolve, reject) => {
       const { server } = this
       if (!server) {
         resolve(undefined)
         return
       }
-
       server.close((err) => err ? reject(err) : resolve(undefined))
     })
   }

src/config.js

@@ -1,3 +1,7 @@
+/**
+ * @import { Signal, UploadOptions } from './upload/upload.d.ts'
+ */
+
 /**
  * Resolve collector options from CLI args and environment.
  *
@@ -6,11 +10,13 @@
  *
  * @param {string[]} argv CLI arguments (without node/script name).
  * @param {NodeJS.ProcessEnv} env Environment variables.
- * @returns {{ port?: number, outputDir?: string }} Options for Collector.
+ * @returns {{ port?: number, outputDir?: string, upload?: UploadOptions }} Options for Collector.
  */
 export function resolveOptions(argv, env) {
-  /** @type {{ port?: number, outputDir?: string }} */
+  /** @type {{ port?: number, outputDir?: string, upload?: UploadOptions }} */
   const options = {}
+  /** @type {Partial<UploadOptions>} */
+  const upload = {}
 
   if (env.COLLECTIVUS_PORT) {
     const port = parseInt(env.COLLECTIVUS_PORT, 10)
@@ -20,22 +26,82 @@ export function resolveOptions(argv, env) {
     options.outputDir = env.COLLECTIVUS_OUTPUT_DIR
   }
 
+  if (env.COLLECTIVUS_UPLOAD_BUCKET) upload.bucket = env.COLLECTIVUS_UPLOAD_BUCKET
+  if (env.COLLECTIVUS_UPLOAD_PREFIX) upload.prefix = env.COLLECTIVUS_UPLOAD_PREFIX
+  if (env.COLLECTIVUS_UPLOAD_TIME) upload.time = env.COLLECTIVUS_UPLOAD_TIME
+  if (env.COLLECTIVUS_UPLOAD_SIGNALS) upload.signals = parseSignals(env.COLLECTIVUS_UPLOAD_SIGNALS)
+  if (env.COLLECTIVUS_UPLOAD_CATCHUP_DAYS) {
+    const n = parseInt(env.COLLECTIVUS_UPLOAD_CATCHUP_DAYS, 10)
+    if (!Number.isNaN(n)) upload.catchupDays = n
+  }
+  if (env.AWS_REGION) upload.region = env.AWS_REGION
+  if (env.COLLECTIVUS_UPLOAD_ENDPOINT) upload.endpoint = env.COLLECTIVUS_UPLOAD_ENDPOINT
+
   for (let i = 0; i < argv.length; i++) {
     const arg = argv[i]
-    if (arg === '--port' && argv[i + 1]) {
-      const port = parseInt(argv[i + 1], 10)
+    function next() { return argv[++i] }
+    if (arg === '--port' && argv[i + 1] !== undefined) {
+      const port = parseInt(next(), 10)
       if (!Number.isNaN(port)) options.port = port
-      i++
     } else if (arg.startsWith('--port=')) {
       const port = parseInt(arg.slice('--port='.length), 10)
       if (!Number.isNaN(port)) options.port = port
-    } else if (arg === '--output' && argv[i + 1]) {
-      options.outputDir = argv[i + 1]
-      i++
+    } else if (arg === '--output' && argv[i + 1] !== undefined) {
+      options.outputDir = next()
     } else if (arg.startsWith('--output=')) {
       options.outputDir = arg.slice('--output='.length)
+    } else if (arg === '--upload-bucket' && argv[i + 1] !== undefined) {
+      upload.bucket = next()
+    } else if (arg.startsWith('--upload-bucket=')) {
+      upload.bucket = arg.slice('--upload-bucket='.length)
+    } else if (arg === '--upload-prefix' && argv[i + 1] !== undefined) {
+      upload.prefix = next()
+    } else if (arg.startsWith('--upload-prefix=')) {
+      upload.prefix = arg.slice('--upload-prefix='.length)
+    } else if (arg === '--upload-time' && argv[i + 1] !== undefined) {
+      upload.time = next()
+    } else if (arg.startsWith('--upload-time=')) {
+      upload.time = arg.slice('--upload-time='.length)
+    } else if (arg === '--upload-signals' && argv[i + 1] !== undefined) {
+      upload.signals = parseSignals(next())
+    } else if (arg.startsWith('--upload-signals=')) {
+      upload.signals = parseSignals(arg.slice('--upload-signals='.length))
+    } else if (arg === '--upload-catchup-days' && argv[i + 1] !== undefined) {
+      const n = parseInt(next(), 10)
+      if (!Number.isNaN(n)) upload.catchupDays = n
+    } else if (arg.startsWith('--upload-catchup-days=')) {
+      const n = parseInt(arg.slice('--upload-catchup-days='.length), 10)
+      if (!Number.isNaN(n)) upload.catchupDays = n
+    } else if (arg === '--upload-region' && argv[i + 1] !== undefined) {
+      upload.region = next()
+    } else if (arg.startsWith('--upload-region=')) {
+      upload.region = arg.slice('--upload-region='.length)
+    } else if (arg === '--upload-endpoint' && argv[i + 1] !== undefined) {
+      upload.endpoint = next()
+    } else if (arg.startsWith('--upload-endpoint=')) {
+      upload.endpoint = arg.slice('--upload-endpoint='.length)
     }
   }
 
+  if (upload.bucket) {
+    options.upload = /** @type {UploadOptions} */ (upload)
+  }
+
   return options
 }
+
+/**
+ * @param {string} value comma-separated signal list
+ * @returns {ReadonlyArray<Signal>}
+ */
+function parseSignals(value) {
+  /** @type {Signal[]} */
+  const out = []
+  for (const part of value.split(',')) {
+    const trimmed = part.trim()
+    if (trimmed === 'logs' || trimmed === 'traces' || trimmed === 'metrics') {
+      out.push(trimmed)
+    }
+  }
+  return out
+}

src/index.js

@@ -1 +1,2 @@
 export { Collector } from './collector.js'
+export { createUploader } from './upload/index.js'

src/upload/connectors/memory.js

@@ -0,0 +1,27 @@
+/**
+ * @import { StorageConnector } from '../upload.d.ts'
+ */
+
+/**
+ * In-memory StorageConnector for tests. Backed by a Map<string, Uint8Array>
+ * exposed on `connector.store` so tests can introspect what was uploaded.
+ *
+ * @returns {StorageConnector & { store: Map<string, Uint8Array> }}
+ */
+export function memoryConnector() {
+  /** @type {Map<string, Uint8Array>} */
+  const store = new Map()
+
+  return {
+    scheme: 'memory',
+    store,
+    async putObject(key, body) {
+      store.set(key, body)
+    },
+    async headObject(key) {
+      const body = store.get(key)
+      if (!body) return null
+      return { size: body.byteLength }
+    },
+  }
+}