Orchestrator: bug fix for logging for workflows feature (#2472) (#2522)

* fix: RHDHBUGS-2735 - Logging For Workflows  (#2411)

* fix: Fixes RHDHBUGS-2735. Properly adds the auth token to the loki query request

* chore: update the changeset to be a patch release (#2430)

---------

Co-authored-by: Hope Hadfield <hhadfiel@redhat.com>
Co-authored-by: Lucas Holmquist <lholmqui@redhat.com>

Author: 3 people

workspaces/orchestrator/.changeset/light-spiders-grin.md

@@ -0,0 +1,7 @@
+---
+'@red-hat-developer-hub/backstage-plugin-orchestrator-backend-module-loki': patch
+---
+
+Fixes https://issues.redhat.com/browse/RHDHBUGS-2735. Adds the ability to pass an auth token to the lokistack query route
+
+Also adds the ability to add custom Loki log pipeline filters

workspaces/orchestrator/plugins/orchestrator-backend-module-loki/README.md

@@ -13,15 +13,24 @@ orchestrator:
   workflowLogProvider:
     loki:
       baseUrl: http://localhost:3100
+      token: secrettoken
+      # rejectUnauthorized: false
+      # logPipelineFilters:
+      # - '| filter1'
+      # - '|= filter2'
       # logStreamSelectors:
       #   - label: 'app'
       #     value: '=~".+"'
 ```
 
-The `baseUrl` is required.
+The `baseUrl` and `token` are required.
+
+If you're baseURL is using a self-signed certificate, add the `rejectUnauthorized: false` parameter. Note: this should only be used in development.
 
 Multiple Log Stream Selectors can be specified in the `logStreamSelectors` section. See the loki docs to learn more about log stream selectors and their values: https://grafana.com/docs/loki/latest/query/log_queries/#log-stream-selector
 
+Multiple Log Pipeline Filters can be specified in the `logPipelineFilters` section. See the loki docs to learn more about the log pipeline filters and their values and usage: https://grafana.com/docs/loki/latest/query/log_queries/#log-pipeline
+
 ## Installation
 
 To install this backend module:

workspaces/orchestrator/plugins/orchestrator-backend-module-loki/config.d.ts

@@ -23,8 +23,22 @@ export interface Config {
       loki?: {
         /**
          * Base URL of the Loki service.
+         * /loki/api/v1/query_range will be appended to the baseUrl
          */
         baseUrl: string;
+        /**
+         * Auth Token for accessing the loki query url
+         */
+        token: string;
+        /**
+         * Set to false if the baseUrl has a self-signed certificate
+         * defaults to true
+         */
+        rejectUnauthorized?: boolean;
+        // Add custom log pipeline filters
+        // default is the workflow instanceId
+        // new values will be appened
+        logPipelineFilters?: Array<string>;
         logStreamSelectors?: Array<{
           // label is the selector, something like 'app' or 'service_name', etc...
           label: string;

workspaces/orchestrator/plugins/orchestrator-backend-module-loki/package.json

@@ -40,7 +40,8 @@
     "@backstage/backend-plugin-api": "^1.5.0",
     "@red-hat-developer-hub/backstage-plugin-orchestrator-common": "workspace:^",
     "@red-hat-developer-hub/backstage-plugin-orchestrator-node": "workspace:^",
-    "luxon": "^3.7.2"
+    "luxon": "^3.7.2",
+    "undici": "^7.22.0"
   },
   "devDependencies": {
     "@backstage/backend-test-utils": "^1.10.0",

workspaces/orchestrator/plugins/orchestrator-backend-module-loki/src/workflowLogsProviders/LokiProvider.test.ts

@@ -27,6 +27,7 @@ describe('LokiProvider', () => {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
             },
           },
         },
@@ -45,13 +46,21 @@ describe('LokiProvider', () => {
       // Test the selectors passed in
       // Should be an empty array when nothing is passed in
       expect(provider.getSelectors()).toEqual([]);
+
+      // Test the token passed in
+      expect(provider.getToken()).toEqual('notsecret');
+
+      // Test the default rejectUnauthorized value
+      expect(provider.getRejectUnauthorized()).toEqual(true);
     });
     it('should create a provider with custom selectors', () => {
       const lokiAppConfig = {
         orchestrator: {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
+              rejectUnauthorized: false,
               logStreamSelectors: [
                 {
                   label: 'custom-selector',
@@ -70,6 +79,9 @@ describe('LokiProvider', () => {
       const lokiConfig = new ConfigReader(lokiAppConfig);
       const provider = LokiProvider.fromConfig(lokiConfig);
 
+      // Test the rejectUnauthorized value when set to false
+      expect(provider.getRejectUnauthorized()).toEqual(false);
+
       // Test the selectors passed in
       expect(provider.getSelectors()[0]).toEqual(
         lokiAppConfig.orchestrator.workflowLogProvider.loki
@@ -98,6 +110,7 @@ describe('LokiProvider', () => {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
             },
           },
         },
@@ -114,12 +127,11 @@ describe('LokiProvider', () => {
       };
 
       const urlToFetch =
-        'http://localhost:3100/loki/api/v1/query_range?query=%7Bservice_name%3D%7E%22.%2B%22%7D+%7C%3D%2212345%22&start=2025-12-05T16%3A30%3A13.621Z&end=2026-01-03T16%3A35%3A13.621Z';
+        'http://localhost:3100/loki/api/v1/query_range?query=%7Bopenshift_log_type%3D%22application%22%7D+%7C%3D%2212345%22&start=2025-12-05T16%3A30%3A13.621Z&end=2026-01-03T16%3A35%3A13.621Z';
       const workflowLogs =
         await provider.fetchWorkflowLogsByInstance(workflowInstance);
 
       const parsedURLToFetch = new URL(urlToFetch);
-      expect(fetch).toHaveBeenCalledWith(urlToFetch);
       expect(parsedURLToFetch.origin).toEqual(provider.getBaseURL());
       expect(parsedURLToFetch.pathname).toEqual('/loki/api/v1/query_range');
       expect(parsedURLToFetch.searchParams.get('end')).toEqual(
@@ -129,7 +141,7 @@ describe('LokiProvider', () => {
         '2025-12-05T16:30:13.621Z',
       ); // should be 5 minutes before
       expect(parsedURLToFetch.searchParams.get('query')).toEqual(
-        `{service_name=~".+"} |="${workflowInstance.id}"`,
+        `{openshift_log_type="application"} |="${workflowInstance.id}"`,
       );
       expect(workflowLogs).toHaveProperty('instanceId', workflowInstance.id);
       expect(workflowLogs).toHaveProperty('logs');
@@ -154,6 +166,7 @@ describe('LokiProvider', () => {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
             },
           },
         },
@@ -174,7 +187,7 @@ describe('LokiProvider', () => {
 
       await provider.fetchWorkflowLogsByInstance(workflowInstance);
       const parsedURLToFetch = new URL(urlToFetch);
-      expect(fetch).toHaveBeenCalledWith(urlToFetch);
+
       expect(parsedURLToFetch.searchParams.get('end')).toEqual(
         '2025-12-05T17:40:13.621Z',
       ); // Should be 5 minutes after
@@ -193,6 +206,7 @@ describe('LokiProvider', () => {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
               logStreamSelectors: [
                 {
                   label: 'custom-selector',
@@ -224,7 +238,7 @@ describe('LokiProvider', () => {
       await provider.fetchWorkflowLogsByInstance(workflowInstance);
 
       const parsedURLToFetch = new URL(urlToFetch);
-      expect(fetch).toHaveBeenCalledWith(urlToFetch);
+
       expect(parsedURLToFetch.origin).toEqual(provider.getBaseURL());
       expect(parsedURLToFetch.pathname).toEqual('/loki/api/v1/query_range');
       expect(parsedURLToFetch.searchParams.get('query')).toEqual(
@@ -245,6 +259,7 @@ describe('LokiProvider', () => {
           workflowLogProvider: {
             loki: {
               baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
               logStreamSelectors: [
                 {
                   value: '=~".+"',
@@ -273,12 +288,55 @@ describe('LokiProvider', () => {
 
       await provider.fetchWorkflowLogsByInstance(workflowInstance);
       const parsedURLToFetch = new URL(urlToFetch);
-      expect(fetch).toHaveBeenCalledWith(urlToFetch);
+
       expect(parsedURLToFetch.origin).toEqual(provider.getBaseURL());
       expect(parsedURLToFetch.pathname).toEqual('/loki/api/v1/query_range');
       expect(parsedURLToFetch.searchParams.get('query')).toEqual(
         `{service_name=~".+",custom-selector1=~".+"} |="${workflowInstance.id}"`,
       );
     });
+
+    it('should have a custom pipeline filter, no label and no value, use defaults', async () => {
+      const mockResponse: Partial<Response> = {
+        ok: true,
+        status: 200,
+        json: jest.fn().mockResolvedValue(mockWorkflowLog),
+      };
+      global.fetch = jest.fn().mockResolvedValue(mockResponse as any);
+
+      const lokiAppConfig = {
+        orchestrator: {
+          workflowLogProvider: {
+            loki: {
+              baseUrl: 'http://localhost:3100',
+              token: 'notsecret',
+              logPipelineFilters: ['| filter1', '| json'],
+            },
+          },
+        },
+      };
+
+      const lokiConfig = new ConfigReader(lokiAppConfig);
+      const provider = LokiProvider.fromConfig(lokiConfig);
+      const workflowInstance: ProcessInstanceDTO = {
+        id: '12345',
+        processId: '54321',
+        start: '2025-12-05T16:35:13.621Z',
+        end: '',
+        nodes: [],
+      };
+
+      const urlToFetch =
+        'http://localhost:3100/loki/api/v1/query_range?query=%7Bopenshift_log_type%3D%22application%22%7D+%7C%3D%2212345%22+%7C+filter1+%7C+json&start=2025-12-05T16%3A30%3A13.621Z&end=2026-01-03T16%3A35%3A13.621Z';
+
+      await provider.fetchWorkflowLogsByInstance(workflowInstance);
+      const parsedURLToFetch = new URL(urlToFetch);
+
+      expect(parsedURLToFetch.origin).toEqual(provider.getBaseURL());
+      expect(parsedURLToFetch.pathname).toEqual('/loki/api/v1/query_range');
+      expect(parsedURLToFetch.searchParams.get('query')).toEqual(
+        `{openshift_log_type="application"} |="${workflowInstance.id}" | filter1 | json`,
+      );
+    });
   });
 });

workspaces/orchestrator/plugins/orchestrator-backend-module-loki/src/workflowLogsProviders/LokiProvider.ts

@@ -22,12 +22,29 @@ import {
 } from '@red-hat-developer-hub/backstage-plugin-orchestrator-common';
 import { WorkflowLogProvider } from '@red-hat-developer-hub/backstage-plugin-orchestrator-node';
 
+import { Agent } from 'undici';
+
+// Augment the global RequestInit interface to include the dispatcher property
+declare global {
+  interface RequestInit {
+    dispatcher?: Agent | undefined;
+  }
+}
+
 export class LokiProvider implements WorkflowLogProvider {
   private readonly baseURL: string;
+  private readonly token: string;
   private readonly selectors: any;
+  private readonly rejectUnauthorized: boolean;
+  private readonly logPipelineFilters: any;
   private constructor(config: Config) {
     this.baseURL = config.getString('baseUrl');
+    this.token = config.getString('token');
+    // Only should be false if specified, undefined here should be true
+    this.rejectUnauthorized =
+      config.getOptionalBoolean('rejectUnauthorized') === false ? false : true;
     this.selectors = config.getOptional('logStreamSelectors') || [];
+    this.logPipelineFilters = config.getOptional('logPipelineFilters') || [];
   }
   getBaseURL(): string {
     return this.baseURL;
@@ -41,6 +58,14 @@ export class LokiProvider implements WorkflowLogProvider {
     return this.selectors;
   }
 
+  getToken(): string {
+    return this.token;
+  }
+
+  getRejectUnauthorized(): boolean {
+    return this.rejectUnauthorized;
+  }
+
   async fetchWorkflowLogsByInstance(
     instance: ProcessInstanceDTO,
   ): Promise<WorkflowLogsResponse> {
@@ -63,13 +88,14 @@ export class LokiProvider implements WorkflowLogProvider {
     const lokiApiEndpoint = '/loki/api/v1/query_range';
     // Query is created with a log stream selector and then a log pipeline for more filtering
     // format looks like this: {stream-selector=expression} | log pipeline/log filter expression
-    // The log stream selector part of the query here is getting all service names
+    // The log stream selector part of the query here is defaulting to openshift_log_type=application
+    // This is the value used for Openshift Logging
     // This might need to be configurable, based on https://grafana.com/docs/loki/latest/query/log_queries/#log-stream-selector
     // Log pipeline part looks for the workflow instance id in those logs
     // Create the streamSelector
     let streamSelector: string = '';
     if (this.selectors.length < 1) {
-      streamSelector = 'service_name=~".+"';
+      streamSelector = 'openshift_log_type="application"';
     } else {
       this.selectors.forEach(
         (
@@ -78,11 +104,18 @@ export class LokiProvider implements WorkflowLogProvider {
           arr: string | any[],
         ) => {
           // something about that last comma
-          streamSelector += `${entry.label || 'service_name'}${entry.value || '=~".+"'}${index !== arr.length - 1 ? ',' : ''}`;
+          streamSelector += `${entry.label || 'openshift_log_type'}${entry.value || '=~".+"'}${index !== arr.length - 1 ? ',' : ''}`;
         },
       );
     }
-    const logPipelineFilter = `|="${instance.id}"`;
+    let logPipelineFilter: string = `|="${instance.id}"`;
+
+    if (this.logPipelineFilters.length > 0) {
+      this.logPipelineFilters.forEach((element: any) => {
+        logPipelineFilter += ` ${element}`;
+      });
+    }
+
     const params = new URLSearchParams({
       query: `{${streamSelector}} ${logPipelineFilter}`,
       start: startTime as string,
@@ -91,9 +124,22 @@ export class LokiProvider implements WorkflowLogProvider {
 
     const urlToFetch = `${this.baseURL}${lokiApiEndpoint}?${params.toString()}`;
 
+    const customAgent = new Agent({
+      connect: {
+        // Agent configuration options, e.g., for self-signed certificates
+        rejectUnauthorized: this.rejectUnauthorized,
+      },
+    });
+
     let allResults;
     try {
-      const response = await fetch(urlToFetch);
+      const response = await fetch(urlToFetch, {
+        dispatcher: customAgent,
+        headers: {
+          Authorization: `Bearer ${this.token}`,
+          'Content-Type': 'application/json',
+        },
+      });
       if (!response.ok) {
         throw new Error(await response.text());
       }

workspaces/orchestrator/yarn.lock

@@ -12427,6 +12427,7 @@ __metadata:
     "@red-hat-developer-hub/backstage-plugin-orchestrator-common": "workspace:^"
     "@red-hat-developer-hub/backstage-plugin-orchestrator-node": "workspace:^"
     luxon: ^3.7.2
+    undici: ^7.22.0
   languageName: unknown
   linkType: soft
 
@@ -37075,10 +37076,10 @@ __metadata:
   languageName: node
   linkType: hard
 
-"undici@npm:^7.2.3":
-  version: 7.3.0
-  resolution: "undici@npm:7.3.0"
-  checksum: 5cf37a51d1449370c7c936fdfefb6be907a59af2829520475a837e81fe709771cf678428fe6beb4c80b168da0570496d13b2ecbf6f52211f844b2a360c6fc6e0
+"undici@npm:^7.2.3, undici@npm:^7.22.0":
+  version: 7.22.0
+  resolution: "undici@npm:7.22.0"
+  checksum: 3eaad0283f946ce12d4aa70da49d23870fcab9a3de5ae0b2855710b58fece570b1675913268dfa8b97589385ff4fb706c004a0614c3b193b617475e7d0a34e3a
   languageName: node
   linkType: hard