diff --git a/.dockerignore b/.dockerignore index bfbdee6..7f806c6 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,6 +4,9 @@ tests .github coverage node_modules +docker +scripts +docker-compose.yaml .prettier* commitlint.config.json README.md diff --git a/.github/workflows/jira-integration.yaml b/.github/workflows/jira-integration.yaml new file mode 100644 index 0000000..34d4ee9 --- /dev/null +++ b/.github/workflows/jira-integration.yaml @@ -0,0 +1,14 @@ +name: Jira Integration +on: + pull_request: + types: [opened, edited, synchronize] + +permissions: + statuses: write + pull-requests: write + +jobs: + jira-validation: + runs-on: ubuntu-latest + steps: + - uses: mapcolonies/javascript-github-actions/actions/jira-integration@jira-integration-v1 diff --git a/README.md b/README.md index 805cd8c..369b3af 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,29 @@ npm run start ``` +## Debugging Locally + +In production, [feedback-api](https://github.com/MapColonies/feedback-api) is the service that produces the Kafka message this service consumes: it combines the feedback it receives (`request_id`, `chosen_response_id`, `user_id`) with Geocoding's original response and publishes the result to the `kafkaTopics.input` topic. `StreamerBuilder` ([src/streamerBuilder.ts](src/streamerBuilder.ts)) is the consumer on the other end — it's the thing being debugged here. + +To exercise that consumer locally without running feedback-api or waiting for real traffic, [scripts/debug-kafka-consumer.ts](scripts/debug-kafka-consumer.ts) stands in for feedback-api: it produces a FeedbackResponse payload, shaped exactly like what feedback-api sends, straight to a local Kafka broker. + +1. Start the dependencies (Kafka, Elasticsearch, the userData mock, and Jaeger for tracing): + ```bash + docker compose up kafka kafka-init elasticsearch user-data-service jaeger + ``` +2. Run the service in dev mode. `npm run start:dev` already points at this docker-compose stack (Kafka's host-reachable `localhost:9094` listener, Elasticsearch on `localhost:9200`, the mock user-data-service on `localhost:5000`, and tracing exported to Jaeger at `localhost:4318`) — see the `start:dev` script in [package.json](package.json) if you need to point it elsewhere: + ```bash + npm run start:dev + ``` +3. Produce a message, standing in for feedback-api: + ```bash + npm run debug:produce-kafka + ``` + +`docker-compose.yaml`'s Kafka broker advertises that host-reachable listener on port `9094` (in addition to the in-network `kafka:9092` one) so the service and the script can run outside docker while talking to it. See [scripts/debug-kafka-consumer.ts](scripts/debug-kafka-consumer.ts) for passing a custom payload file (e.g. one captured from production) instead of the bundled sample, or `--new-request-id` to replay the same payload without colliding on request ids. + +Once consumed, the enriched record lands in Elasticsearch (`curl localhost:9200/enrich_index/_search`) and the full trace — Kafka handling, the user-data lookup, and the Elasticsearch write — is viewable in the Jaeger UI at http://localhost:16686. + ## Running Tests To run tests, run the following command @@ -109,3 +132,17 @@ To only run integration tests: ```bash npm run test:integration ``` + +## Deploy to azure + +```bash +docker build . --tag acrarolibotnonprod.azurecr.io/geocoding-enrichment:v1.7.0-rc2 +docker push acrarolibotnonprod.azurecr.io/geocoding-enrichment:v1.7.0-rc2 +``` + +go to https://github.com/MapColonies/vector-helms/tree/master/geocoding-helms/charts/geocoding-enrichment +update image in values and run: + +``` +helm upgrade --install vector-geocoding -f azure.dev.yaml . -n vector-dev +``` diff --git a/config/custom-environment-variables.json b/config/custom-environment-variables.json index 02d41be..17634bc 100644 --- a/config/custom-environment-variables.json +++ b/config/custom-environment-variables.json @@ -49,7 +49,7 @@ "sslPaths": { "ca": "KAFKA_CA_PATH", "key": "KAFKA_KEY_PATH", - "cert": "KAFKA_CERT_PATkafH" + "cert": "KAFKA_CERT_PATH" }, "sasl": { "mechanism": "KAFKA_SASL_MECHANISM", diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..75ad91e --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,74 @@ +services: + kafka: + image: apache/kafka:3.9.1 + environment: + - KAFKA_NODE_ID=1 + - KAFKA_PROCESS_ROLES=broker,controller + - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:9094 + - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,HOST://localhost:9094 + - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER + - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT + - KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 + - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 + - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 + - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 + - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 + - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true + ports: + - '9092:9092' + - '9094:9094' # host-reachable listener, so the service/scripts can run outside docker while it talks to this broker + volumes: + - kafka-data:/var/lib/kafka/data + healthcheck: + test: ['CMD-SHELL', '/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092'] + interval: 10s + timeout: 10s + retries: 15 + + elasticsearch: + image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0 + environment: + - discovery.type=single-node + - xpack.security.enabled=false + - ES_JAVA_OPTS=-Xms512m -Xmx512m + ports: + - '9200:9200' + volumes: + - elasticsearch-data:/usr/share/elasticsearch/data + healthcheck: + test: ['CMD-SHELL', 'curl -s http://localhost:9200/_cluster/health | grep -q ''"status"'''] + interval: 10s + timeout: 5s + retries: 15 + + kafka-init: + image: apache/kafka:3.9.1 + depends_on: + kafka: + condition: service_healthy + entrypoint: ['/bin/sh', '-c'] + command: + - '/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists --topic topic --partitions 1 --replication-factor 1' + restart: 'no' + + user-data-service: + image: node:20-alpine + working_dir: /usr/src/app + volumes: + - .:/usr/src/app:ro + - /usr/src/app/dist + command: ['sh', '-c', 'npx tsc --project tsconfig.json --skipLibCheck && node dist/docker/user-data-service/server.js'] + ports: + - '5000:8080' + + jaeger: + image: jaegertracing/all-in-one:1.62.0 + environment: + - COLLECTOR_OTLP_ENABLED=true + ports: + - '16686:16686' # web UI + - '4318:4318' # OTLP HTTP receiver + +volumes: + kafka-data: + elasticsearch-data: diff --git a/docker/user-data-service/server.ts b/docker/user-data-service/server.ts new file mode 100644 index 0000000..84c5e5c --- /dev/null +++ b/docker/user-data-service/server.ts @@ -0,0 +1,41 @@ +import { createServer, IncomingMessage, ServerResponse } from 'http'; +import httpStatus from 'http-status-codes'; +import { UserDataServiceResponse } from '../../src/common/interfaces'; + +const port = 8080; + +// eslint-disable-next-line @typescript-eslint/naming-convention +const jsonHeaders = { 'Content-Type': 'application/json' }; + +/* eslint-disable @typescript-eslint/naming-convention */ +const users: UserDataServiceResponse = { + 'avi@mapcolonies.net': { + firstName: 'avi', + lastName: 'map', + displayName: 'mapcolonies/avi', + mail: 'avi@mapcolonies.net', + domains: ['USA', 'FRANCE'], + }, +}; +/* eslint-enable @typescript-eslint/naming-convention */ + +const server = createServer((req: IncomingMessage, res: ServerResponse) => { + const match = /^\/user_data\/([^/?]+)/.exec(req.url ?? ''); + + if (req.method !== 'GET' || !match) { + res.writeHead(httpStatus.NOT_FOUND, jsonHeaders); + res.end(JSON.stringify({ message: 'not found' })); + return; + } + + const userId = decodeURIComponent(match[1]); + console.log('new request', { userid: userId }); + const user = users[userId] ?? { [userId]: null }; + + res.writeHead(httpStatus.OK, jsonHeaders); + res.end(JSON.stringify(user)); +}); + +server.listen(port, () => { + console.log(`user-data-service mock is running on port ${port}`); +}); diff --git a/package-lock.json b/package-lock.json index 1f1cfd7..df4a481 100644 --- a/package-lock.json +++ b/package-lock.json @@ -50,6 +50,7 @@ "@types/jest": "^29.5.2", "@types/jsonwebtoken": "^9.0.6", "@types/multer": "^1.4.7", + "@types/node": "^17.0.36", "@types/supertest": "^2.0.12", "@types/swagger-ui-express": "^4.1.3", "commitlint": "^17.6.6", @@ -6677,7 +6678,8 @@ "node_modules/@types/node": { "version": "17.0.36", "resolved": "https://registry.npmjs.org/@types/node/-/node-17.0.36.tgz", - "integrity": "sha512-V3orv+ggDsWVHP99K3JlwtH20R7J4IhI1Kksgc+64q5VxgfRkQG8Ws3MFm/FZOKDYGy9feGFlZ70/HpCNe9QaA==" + "integrity": "sha512-V3orv+ggDsWVHP99K3JlwtH20R7J4IhI1Kksgc+64q5VxgfRkQG8Ws3MFm/FZOKDYGy9feGFlZ70/HpCNe9QaA==", + "license": "MIT" }, "node_modules/@types/normalize-package-data": { "version": "2.4.1", @@ -17980,6 +17982,7 @@ "resolved": "https://registry.npmjs.org/ts-node/-/ts-node-10.9.1.tgz", "integrity": "sha512-NtVysVPkxxrwFGUUxGYhfux8k78pQB3JqYBXlLRZgdGUqTO5wU/UyHop5p70iEbGhB7q5KmiZiU0Y3KlJrScEw==", "dev": true, + "license": "MIT", "dependencies": { "@cspotcode/source-map-support": "^0.8.0", "@tsconfig/node10": "^1.0.7", diff --git a/package.json b/package.json index 2014baf..b8da2da 100644 --- a/package.json +++ b/package.json @@ -18,9 +18,10 @@ "prebuild": "npm run clean", "build": "tsc --project tsconfig.build.json && npm run assets:copy", "start": "npm run build && cd dist && node --require ./common/tracing.js ./index.js", - "start:dev": "npm run build && cd dist && node --enable-source-maps --require ./common/tracing.js ./index.js", + "start:dev": "npm run build && cd dist && KAFKA_BROKERS=localhost:9094 ELASTIC_URL=http://localhost:9200 USER_DATA_SERVICE_ENDPOINT=http://localhost:5000/user_data TELEMETRY_TRACING_ENABLED=true TELEMETRY_TRACING_URL=http://localhost:4318/v1/traces node --enable-source-maps --require ./common/tracing.js ./index.js", "assets:copy": "copyfiles -f ./config/* ./dist/config && copyfiles -f ./openapi3.yaml ./dist/ && copyfiles ./package.json dist", "clean": "rimraf dist", + "debug:produce-kafka": "tsc --project tsconfig.json --skipLibCheck && node dist/scripts/debug-kafka-consumer.js", "install": "npx husky install" }, "directories": { @@ -80,6 +81,7 @@ "@types/jest": "^29.5.2", "@types/jsonwebtoken": "^9.0.6", "@types/multer": "^1.4.7", + "@types/node": "^17.0.36", "@types/supertest": "^2.0.12", "@types/swagger-ui-express": "^4.1.3", "commitlint": "^17.6.6", diff --git a/scripts/debug-kafka-consumer.ts b/scripts/debug-kafka-consumer.ts new file mode 100755 index 0000000..2ce312e --- /dev/null +++ b/scripts/debug-kafka-consumer.ts @@ -0,0 +1,62 @@ +// Run via `npm run debug:produce-kafka`. See README.md "Debugging Locally" for usage, prerequisites, and env vars. + +import { readFileSync } from 'fs'; +import { join } from 'path'; +import { Kafka } from 'kafkajs'; +import { FeedbackResponse } from '../src/common/interfaces'; + +const DEFAULT_PAYLOAD_PATH = join(process.cwd(), 'scripts', 'sample-payloads', 'feedback-response.json'); +const CLI_ARGS_OFFSET = 2; // skip `node` and the script path in argv + +interface ParsedArgs { + payloadPath: string; + newRequestId: boolean; +} + +function parseArgs(argv: string[]): ParsedArgs { + const positional = argv.filter((arg) => !arg.startsWith('--')); + return { + payloadPath: positional[0] ?? DEFAULT_PAYLOAD_PATH, + newRequestId: argv.includes('--new-request-id'), + }; +} + +function loadPayload(payloadPath: string, newRequestId: boolean): FeedbackResponse { + const raw = readFileSync(payloadPath, 'utf-8'); + const payload = JSON.parse(raw) as FeedbackResponse; + + if (newRequestId) { + payload.requestId = `debug-${process.pid}-${process.hrtime.bigint()}`; + } + + return payload; +} + +async function main(): Promise { + const { payloadPath, newRequestId } = parseArgs(process.argv.slice(CLI_ARGS_OFFSET)); + const payload = loadPayload(payloadPath, newRequestId); + + const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','); + const topic = process.env.KAFKA_TOPIC ?? 'topic'; + + const kafka = new Kafka({ clientId: 'debug-kafka-producer', brokers }); + const producer = kafka.producer(); + + await producer.connect(); + try { + const result = await producer.send({ + topic, + messages: [{ value: JSON.stringify(payload) }], + }); + + console.log(`Sent ${payloadPath} to topic "${topic}" on [${brokers.join(', ')}] (requestId: ${payload.requestId})`); + console.log(result); + } finally { + await producer.disconnect(); + } +} + +main().catch((error: unknown) => { + console.error('Failed to produce debug message:', error); + process.exitCode = 1; +}); diff --git a/scripts/sample-payloads/feedback-response.json b/scripts/sample-payloads/feedback-response.json new file mode 100644 index 0000000..15d7e2b --- /dev/null +++ b/scripts/sample-payloads/feedback-response.json @@ -0,0 +1,44 @@ +{ + "requestId": "debug-local-001", + "chosenResultId": 0, + "responseTime": "2026-10-05T10:00:03.500Z", + "geocodingResponse": { + "userId": "avi@mapcolonies.net", + "apiKey": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJtYXAtY29sb25pZXMtdGVzdCIsIm5hbWUiOiJKb2huIERvZSIsImlhdCI6MTUxNjIzOTAyMn0.cyKspTifStJI-w2lYk05QaFNyyHYDtUupJddqVj7ZS8", + "site": "geocoding-web-client", + "respondedAt": "2026-10-05T10:00:03.000Z", + "response": { + "type": "FeatureCollection", + "geocoding": { + "version": "1.0", + "query": { + "query": "דיזנגוף 99, תל אביב" + } + }, + "features": [ + { + "type": "Feature", + "geometry": { "type": "Point", "coordinates": [34.774302, 32.084041] }, + "properties": { + "type": "Point", + "matches": [{ "layer": "address", "source": "osm", "source_id": ["way/123456"] }], + "names": { "default": "דיזנגוף 99, תל אביב - יפו" }, + "regions": [{ "region": "תל אביב - יפו", "sub_region_names": ["מרכז"] }], + "_score": 0.97 + } + }, + { + "type": "Feature", + "geometry": { "type": "Point", "coordinates": [34.781, 32.09] }, + "properties": { + "type": "Point", + "matches": [{ "layer": "address", "source": "osm", "source_id": ["way/654321"] }], + "names": { "default": "דיזנגוף 101, תל אביב - יפו" }, + "regions": [{ "region": "תל אביב - יפו", "sub_region_names": ["מרכז"] }], + "_score": 0.81 + } + } + ] + } + } +} diff --git a/src/common/tracing.ts b/src/common/tracing.ts index d51d6b6..3444575 100644 --- a/src/common/tracing.ts +++ b/src/common/tracing.ts @@ -1,5 +1,8 @@ -import { Tracing } from '@map-colonies/telemetry'; -import { IGNORED_INCOMING_TRACE_ROUTES, IGNORED_OUTGOING_TRACE_ROUTES } from './constants'; +import { asyncCallWithSpan, callWithSpan, Tracing } from '@map-colonies/telemetry'; +import { trace, type Span, type SpanOptions, type Tracer } from '@opentelemetry/api'; +import { IGNORED_INCOMING_TRACE_ROUTES, IGNORED_OUTGOING_TRACE_ROUTES, SERVICE_NAME } from './constants'; + +const getServiceTracer = (): Tracer => trace.getTracer(SERVICE_NAME); const tracing = new Tracing(undefined, { // eslint-disable-next-line @typescript-eslint/naming-convention @@ -18,3 +21,15 @@ const tracing = new Tracing(undefined, { tracing.start(); export { tracing }; + +/** + * Wraps an async function with a span. When tracing is disabled, fn runs unchanged and span is undefined. + */ +export const withSpan = async (spanName: string, spanOptions: SpanOptions, fn: (span?: Span) => Promise): Promise => + asyncCallWithSpan(fn, getServiceTracer(), spanName, spanOptions); + +/** + * Wraps a sync function with a span. When tracing is disabled, fn runs unchanged and span is undefined. + */ +export const withSpanSync = (spanName: string, spanOptions: SpanOptions, fn: (span?: Span) => T): T => + callWithSpan(fn, getServiceTracer(), spanName, spanOptions); diff --git a/src/process/models/processManager.ts b/src/process/models/processManager.ts index 3e0e4aa..dffcee3 100644 --- a/src/process/models/processManager.ts +++ b/src/process/models/processManager.ts @@ -2,8 +2,10 @@ import { Logger } from '@map-colonies/js-logger'; import { center } from '@turf/center'; import { inject, injectable } from 'tsyringe'; import { SERVICES } from '../../common/constants'; -import { EnrichResponse, FeedbackResponse, IApplication } from '../../common/interfaces'; +import { withSpan } from '../../common/tracing'; +import { EnrichResponse, FeedbackResponse, IApplication, UserDataServiceResponse } from '../../common/interfaces'; import { fetchUserDataService } from '../../common/utils'; +import { ProcessSpanName, ProcessAttributes } from '../tracing'; const arabicRegex = /[\u0600-\u06FF]/; @@ -15,29 +17,41 @@ export class ProcessManager { ) {} public async process(feedbackResponse: FeedbackResponse): Promise { - const token = JSON.parse(Buffer.from(feedbackResponse.geocodingResponse.apiKey.split('.')[1], 'base64').toString()) as { sub: string }; - const text = this.getQueryText(feedbackResponse); - - const enrichedResponse: EnrichResponse = { - query: { - language: arabicRegex.test(text) ? 'ar' : 'he', - text, - }, - result: { - rank: null, + return withSpan( + ProcessSpanName.MANAGER_PROCESS, + { + attributes: { + [ProcessAttributes.SITE]: feedbackResponse.geocodingResponse.site, + [ProcessAttributes.CHOSEN_RESULT_ID]: feedbackResponse.chosenResultId ?? undefined, + }, }, - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - system: token?.sub, - site: feedbackResponse.geocodingResponse.site, - // eslint-disable-next-line @typescript-eslint/ban-ts-comment - // @ts-expect-error - duration: new Date(feedbackResponse.responseTime) - new Date(feedbackResponse.geocodingResponse.respondedAt), - timestamp: new Date(), - }; - if (feedbackResponse.chosenResultId === null) { - return enrichedResponse; - } - return this.enrichData(feedbackResponse, enrichedResponse); + async (span) => { + const token = JSON.parse(Buffer.from(feedbackResponse.geocodingResponse.apiKey.split('.')[1], 'base64').toString()) as { sub: string }; + const text = this.getQueryText(feedbackResponse); + + const enrichedResponse: EnrichResponse = { + query: { + language: arabicRegex.test(text) ? 'ar' : 'he', + text, + }, + result: { + rank: null, + }, + system: token.sub, + site: feedbackResponse.geocodingResponse.site, + duration: new Date(feedbackResponse.responseTime).getTime() - new Date(feedbackResponse.geocodingResponse.respondedAt).getTime(), + timestamp: new Date(), + }; + + span?.setAttributes({ + [ProcessAttributes.LANGUAGE]: enrichedResponse.query.language, + [ProcessAttributes.SYSTEM]: enrichedResponse.system, + [ProcessAttributes.DURATION_MS]: enrichedResponse.duration, + }); + + return this.enrichData(feedbackResponse, enrichedResponse); + } + ); } public getQueryText(feedbackResponse: FeedbackResponse): string { @@ -52,33 +66,58 @@ export class ProcessManager { } public async enrichData(feedbackResponse: FeedbackResponse, enrichedResponse: EnrichResponse): Promise { - const chosenResult: number = feedbackResponse.chosenResultId as number; - const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult]; + const chosenResult = feedbackResponse.chosenResultId; + if (chosenResult === null) { + return enrichedResponse; + } + + return withSpan(ProcessSpanName.MANAGER_ENRICH_DATA, { attributes: { [ProcessAttributes.CHOSEN_RESULT_ID]: chosenResult } }, async (span) => { + const selectedResponse = feedbackResponse.geocodingResponse.response.features[chosenResult]; + + const { endpoint, queryParams, headers } = this.appConfig.userDataService; + const userId = feedbackResponse.geocodingResponse.userId; - const { endpoint, queryParams, headers } = this.appConfig.userDataService; - const userId = feedbackResponse.geocodingResponse.userId; + if (userId !== undefined) { + let fetchedUserData: UserDataServiceResponse; + try { + fetchedUserData = await withSpan( + ProcessSpanName.MANAGER_FETCH_USER_DATA, + { attributes: { [ProcessAttributes.USER_ID]: userId } }, + async () => fetchUserDataService(endpoint, userId, queryParams, headers) + ); + } catch (error) { + this.logger.error(`Failed to fetch user data for user: ${userId}`, error); + throw error; + } - if (userId !== undefined) { - const fetchedUserData = await fetchUserDataService(endpoint, userId, queryParams, headers); + enrichedResponse.user = { + name: userId, + }; - enrichedResponse.user = { - name: userId, + const userData = fetchedUserData[userId] ?? fetchedUserData; + enrichedResponse.user = { ...enrichedResponse.user, ...userData }; + } + + enrichedResponse.result = { + rank: chosenResult, + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + score: selectedResponse.properties?._score ?? 0, + source: selectedResponse.properties.matches[0].source, + layer: selectedResponse.properties.matches[0].layer, + name: selectedResponse.properties.names.default, + region: selectedResponse.properties.regions[0].region, + location: center(selectedResponse), }; - const userData = fetchedUserData[userId] ?? fetchedUserData; - enrichedResponse.user = { ...enrichedResponse.user, ...userData }; - } + span?.setAttributes({ + [ProcessAttributes.RESULT_RANK]: chosenResult, + [ProcessAttributes.RESULT_SCORE]: enrichedResponse.result.score, + [ProcessAttributes.RESULT_SOURCE]: enrichedResponse.result.source, + [ProcessAttributes.RESULT_LAYER]: enrichedResponse.result.layer, + [ProcessAttributes.RESULT_REGION]: enrichedResponse.result.region, + }); - enrichedResponse.result = { - rank: chosenResult, - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - score: selectedResponse.properties?._score ?? 0, - source: selectedResponse.properties.matches[0].source, - layer: selectedResponse.properties.matches[0].layer, - name: selectedResponse.properties.names.default, - region: selectedResponse.properties.regions[0].region, - location: center(selectedResponse), - }; - return enrichedResponse; + return enrichedResponse; + }); } } diff --git a/src/process/tracing.ts b/src/process/tracing.ts new file mode 100644 index 0000000..cc6105b --- /dev/null +++ b/src/process/tracing.ts @@ -0,0 +1,24 @@ +/* eslint-disable @typescript-eslint/naming-convention */ +export const ProcessSpanName = { + MANAGER_PROCESS: 'process.manager.process', + MANAGER_ENRICH_DATA: 'process.manager.enrich_data', + MANAGER_FETCH_USER_DATA: 'process.manager.fetch_user_data', +} as const; + +export type ProcessSpanName = (typeof ProcessSpanName)[keyof typeof ProcessSpanName]; + +export const ProcessAttributes = { + SITE: 'process.site', + LANGUAGE: 'process.query.language', + SYSTEM: 'process.system', + DURATION_MS: 'process.duration_ms', + CHOSEN_RESULT_ID: 'process.chosen_result_id', + USER_ID: 'process.user.id', + RESULT_RANK: 'process.result.rank', + RESULT_SCORE: 'process.result.score', + RESULT_SOURCE: 'process.result.source', + RESULT_LAYER: 'process.result.layer', + RESULT_REGION: 'process.result.region', +} as const; + +export type ProcessAttributes = (typeof ProcessAttributes)[keyof typeof ProcessAttributes]; diff --git a/src/streamerBuilder.ts b/src/streamerBuilder.ts index eedfbc1..2a13059 100644 --- a/src/streamerBuilder.ts +++ b/src/streamerBuilder.ts @@ -5,8 +5,10 @@ import { Client, ClientOptions } from '@elastic/elasticsearch'; import { Consumer, ConsumerConfig, Kafka } from 'kafkajs'; import { CleanupRegistry } from '@map-colonies/cleanup-registry'; import { SERVICES } from './common/constants'; +import { withSpan } from './common/tracing'; import { FeedbackResponse, IConfig, KafkaOptions } from './common/interfaces'; import { ProcessManager } from './process/models/processManager'; +import { StreamerAttributes, StreamerSpanName } from './streamerTracing'; interface KafkaTopics { input: string[]; @@ -53,27 +55,40 @@ export class StreamerBuilder { public async build(): Promise { const { input: inputTopic } = this.config.get('kafkaTopics'); - await this.consumer.connect(); - await this.consumer.subscribe({ topics: inputTopic }); + await withSpan(StreamerSpanName.KAFKA_STARTUP, { attributes: { [StreamerAttributes.KAFKA_TOPIC]: inputTopic.join(',') } }, async () => { + await this.consumer.connect(); + await this.consumer.subscribe({ topics: inputTopic }); + }); this.logger.info(`Kafka consumer subscribed successfully to ${inputTopic.toString()}`); await this.consumer.run({ - eachMessage: async ({ message }) => { + eachMessage: async ({ topic, partition, message }) => { const value = message.value?.toString(); - if (value != undefined) { - try { - const input = JSON.parse(value) as FeedbackResponse; - const requestId = input.requestId; - const output = await this.manager.process(input); + if (value == undefined) { + return; + } - await this.elasticClient.index({ - index: this.config.get(elasticIndex), - body: output, - }); - this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`); - } catch (error) { - this.logger.error(`Error: Could not add data to elastic. Reason: ${(error as Error).message}`); - } + try { + await withSpan( + StreamerSpanName.KAFKA_HANDLE_MESSAGE, + { + attributes: { + [StreamerAttributes.KAFKA_TOPIC]: topic, + [StreamerAttributes.KAFKA_PARTITION]: partition, + [StreamerAttributes.KAFKA_OFFSET]: message.offset, + }, + }, + async () => { + const input = JSON.parse(value) as FeedbackResponse; + const requestId = input.requestId; + const output = await this.manager.process(input); + await this.elasticClient.index({ index: this.config.get(elasticIndex), body: output }); + + this.logger.info(`Added the enriched data of request: ${requestId} to Elastic successfully`); + } + ); + } catch (error) { + this.logger.error(`Error: Could not add data to elastic. Reason: ${(error as Error).message}`); } }, }); diff --git a/src/streamerTracing.ts b/src/streamerTracing.ts new file mode 100644 index 0000000..505ad47 --- /dev/null +++ b/src/streamerTracing.ts @@ -0,0 +1,15 @@ +/* eslint-disable @typescript-eslint/naming-convention */ +export const StreamerSpanName = { + KAFKA_STARTUP: 'streamer.kafka.startup', + KAFKA_HANDLE_MESSAGE: 'streamer.kafka.handle_message', +} as const; + +export type StreamerSpanName = (typeof StreamerSpanName)[keyof typeof StreamerSpanName]; + +export const StreamerAttributes = { + KAFKA_TOPIC: 'streamer.kafka.topic', + KAFKA_PARTITION: 'streamer.kafka.partition', + KAFKA_OFFSET: 'streamer.kafka.offset', +} as const; + +export type StreamerAttributes = (typeof StreamerAttributes)[keyof typeof StreamerAttributes]; diff --git a/tsconfig.json b/tsconfig.json index f5df946..bc2fdd7 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -11,6 +11,6 @@ "experimentalDecorators": true, "emitDecoratorMetadata": true }, - "include": ["src"], + "include": ["src", "scripts", "docker"], "exclude": ["node_modules", "dist", "tests"] }