pipe device log messages to application insights

Author: pelikhan

packages/gateway/src/appinsights.ts

@@ -1,7 +1,7 @@
 import * as mq from "./mq"
 import * as appInsights from "applicationinsights"
 import { ContextTagKeys } from "applicationinsights/out/Declarations/Contracts"
-import { registerMessageSink } from "./messages"
+import { registerLogSink, registerMessageSink } from "./messages"
 
 // telemetry from devices
 let _devsTelemetry: appInsights.TelemetryClient
@@ -14,12 +14,12 @@ export async function setup() {
 
     appInsights
         .setup(connString)
-        .setAutoDependencyCorrelation(true)
-        .setAutoCollectRequests(true)
-        .setAutoCollectPerformance(true, true)
+        .setAutoDependencyCorrelation(false)
+        .setAutoCollectRequests(false)
+        .setAutoCollectPerformance(false)
         .setAutoCollectExceptions(true)
-        .setAutoCollectDependencies(true)
-        .setAutoCollectConsole(true, true)
+        .setAutoCollectDependencies(false)
+        .setAutoCollectConsole(false)
         .setUseDiskRetryCaching(true)
         .setAutoCollectPreAggregatedMetrics(true)
         .setSendLiveMetrics(false)
@@ -34,6 +34,7 @@ export async function setup() {
         name: "server.start",
     })
 
+    registerLogSink(async (logs, device) => device.traceTrace(logs.join("\n")))
     registerMessageSink({
         name: "app insights events",
         topicName: "tev",
@@ -51,7 +52,6 @@ export async function setup() {
             device.trackEvent(`devs.${name}`, { properties, measurements })
         },
     })
-
     registerMessageSink({
         name: "app insights metrics",
         topicName: "tme",

packages/gateway/src/messages.ts

@@ -50,3 +50,12 @@ export async function ingestMessage(
     // dispatch
     await Promise.all(sinks.map(sink => sink.ingest(message, device)))
 }
+
+export type LogSink = (logs: string[], device: ConnectedDevice) => Promise<void>
+const logSinks: LogSink[] = []
+export function registerLogSink(sink: LogSink) {
+    logSinks.push(sink)
+}
+export async function ingestLogs(logs: string[], device: ConnectedDevice) {
+    await Promise.all(logSinks.map(sink => sink(logs, device)))
+}

packages/gateway/src/wssk.ts

@@ -17,9 +17,10 @@ import {
     MetricTelemetry,
     Telemetry,
     TelemetryType,
+    TraceTelemetry,
 } from "applicationinsights/out/Declarations/Contracts"
 import { WsskCmd, WsskDataType } from "./interop"
-import { ingestMessage } from "./messages"
+import { ingestLogs, ingestMessage } from "./messages"
 import { WsskStreamingType } from "../../../devicescript/interop/src/interop"
 
 const JD_AES_CCM_TAG_BYTES = 4
@@ -479,8 +480,12 @@ export class ConnectedDevice {
             case WsskCmd.Dmesg: {
                 const { buf, lines } = splitLines(this.logBuffer, payload)
                 this.logBuffer = buf
-                if (lines.length)
-                    await this.notify({ type: "logs", logs: lines })
+                if (lines.length) {
+                    await Promise.all([
+                        ingestLogs(lines, this),
+                        this.notify({ type: "logs", logs: lines }),
+                    ])
+                }
             }
             default:
                 if (!(await this.deployStep(cmd, payload)))
@@ -565,11 +570,6 @@ export class ConnectedDevice {
     }
 
     public trackMetric(name: string, options: Partial<MetricTelemetry>) {
-        console.log(`metric`, {
-            name,
-            kind: "Aggregation",
-            ...options,
-        })
         this.track(
             {
                 name,
@@ -580,6 +580,17 @@ export class ConnectedDevice {
         )
     }
 
+    public traceTrace(message: string, options: Partial<TraceTelemetry> = {}) {
+        if (!message) return
+        this.track(
+            {
+                message,
+                ...options,
+            },
+            TelemetryType.Trace
+        )
+    }
+
     private trackWarning(message: string) {
         this.trackEvent("warning", { properties: { message } })
     }