PagerDuty: Add option to exclude incidents from specific services (#656)

* Option to exclude services

* Add test and query by service

* Type

sources/pagerduty-source/resources/spec.json

@@ -48,6 +48,18 @@
           "Custom"
         ],
         "pattern": "^(Sev[0-5])?(Custom)?$"
+      },
+      "exclude_services": {
+        "type": "array",
+        "items": {
+          "type": "string"
+        },
+        "title": "Exclude Services",
+        "examples": [
+          "service-1",
+          "service-2"
+        ],
+        "description": "List of PagerDuty service names to ignore incidents from. If not set, all incidents will be pulled."
       }
     }
   }

sources/pagerduty-source/src/pagerduty.ts

@@ -3,7 +3,7 @@ import {AirbyteLogger, wrapApiError} from 'faros-airbyte-cdk';
 import {VError} from 'verror';
 
 export const DEFAULT_CUTOFF_DAYS = 90;
-const DEFAUTL_OVERVIEW = true;
+const DEFAULT_OVERVIEW = true;
 const DEFAULT_PAGE_SIZE = 25; // 25 is API default
 
 enum IncidentSeverityCategory {
@@ -14,7 +14,7 @@ enum IncidentSeverityCategory {
   Sev5 = 'Sev5',
   Custom = 'Custom',
 }
-type IncidentUrgency = 'high' | 'low'; //Pagerduty only has these two priorities
+type IncidentUrgency = 'high' | 'low'; // PagerDuty only has these two priorities
 type IncidentState = 'triggered' | 'acknowledged' | 'resolved';
 
 interface Acknowledgement {
@@ -33,6 +33,7 @@ export interface PagerdutyConfig {
   readonly page_size?: number;
   readonly default_severity?: IncidentSeverityCategory;
   readonly incident_log_entries_overview?: boolean;
+  readonly exclude_services?: ReadonlyArray<string>;
 }
 
 interface PagerdutyResponse<Type> {
@@ -88,6 +89,10 @@ export interface Priority extends PagerdutyObject {
   readonly name: string;
 }
 
+interface Service extends PagerdutyObject {
+  name: string;
+}
+
 export class Pagerduty {
   private static pagerduty: Pagerduty = null;
 
@@ -208,7 +213,7 @@ export class Pagerduty {
 
       timeRange = `&since=${since}&until=${until.toISOString()}`;
     }
-    const limitParam = `&limit=${limit.toFixed()}`;
+    const limitParam = `&limit=${limit}`;
     const teamsResource = `/teams?time_zone=UTC${timeRange}${limitParam}`;
     this.logger.debug(`Fetching Team at ${teamsResource}`);
 
@@ -221,7 +226,8 @@ export class Pagerduty {
 
   async *getIncidents(
     since?: string,
-    limit = DEFAULT_PAGE_SIZE
+    limit = DEFAULT_PAGE_SIZE,
+    exclude_services: ReadonlyArray<string> = []
   ): AsyncGenerator<Incident> {
     let until: Date;
     let timeRange = '&date_range=all';
@@ -232,26 +238,54 @@ export class Pagerduty {
 
       timeRange = `&since=${since}&until=${until.toISOString()}`;
     }
-    const limitParam = `&limit=${limit.toFixed()}`;
-    const incidentsResource = `/incidents?time_zone=UTC${timeRange}${limitParam}`;
-    this.logger.debug(`Fetching Incidents at ${incidentsResource}`);
 
-    const func = (): any => {
-      return this.client.get(incidentsResource);
-    };
+    const limitParam = `&limit=${limit}`;
+    const services: (Service | undefined)[] = [];
+    if (exclude_services?.length > 0) {
+      const servicesIter = this.getServices(limit);
+      for await (const service of servicesIter) {
+        if (
+          exclude_services.includes(service.name) ||
+          exclude_services.includes(service.summary)
+        ) {
+          this.logger.debug(
+            `Excluding Incidents from service id: ${service.id}, name: ${service.name}, summary: ${service.summary}`
+          );
+        } else {
+          services.push(service);
+        }
+      }
+    } else {
+      services.push(undefined); // fetch incidents from all services
+    }
 
-    yield* this.paginate<Incident>(func);
+    // query per service to minimize chance of hitting 10000 records response limit
+    for (const service of services) {
+      let serviceIdsParam = '';
+      if (service) {
+        this.logger.debug(
+          `Fetching Incidents for service id: ${service.id}, name: ${service.name}, summary: ${service.summary}`
+        );
+        serviceIdsParam = `&service_ids[]=${service.id}`;
+      }
+      const incidentsResource = `/incidents?time_zone=UTC${timeRange}${serviceIdsParam}${limitParam}`;
+      this.logger.debug(`Fetching Incidents at ${incidentsResource}`);
+      const func = (): any => {
+        return this.client.get(incidentsResource);
+      };
+      yield* this.paginate<Incident>(func);
+    }
   }
 
   async *getIncidentLogEntries(
     since?: string,
     until?: Date,
     limit: number = DEFAULT_PAGE_SIZE,
-    isOverview = DEFAUTL_OVERVIEW
+    isOverview = DEFAULT_OVERVIEW
   ): AsyncGenerator<LogEntry> {
     const sinceParam = since ? `&since=${since}` : '';
     const untilParam = until ? `&until=${until.toISOString()}` : '';
-    const limitParam = `&limit=${limit.toFixed()}`;
+    const limitParam = `&limit=${limit}`;
     const isOverviewParam = `&is_overview=${isOverview}`;
 
     const logsResource = `/log_entries?time_zone=UTC${sinceParam}${untilParam}${limitParam}${isOverviewParam}`;
@@ -277,4 +311,15 @@ export class Pagerduty {
       }
     }
   }
+
+  async *getServices(
+    limit: number = DEFAULT_PAGE_SIZE
+  ): AsyncGenerator<Service> {
+    const servicesResource = `/services?limit=${limit}`;
+    this.logger.debug(`Fetching Services at ${servicesResource}`);
+    const func = (): any => {
+      return this.client.get(servicesResource);
+    };
+    yield* this.paginate<Service>(func);
+  }
 }

sources/pagerduty-source/src/streams/incidents.ts

@@ -50,7 +50,11 @@ export class Incidents extends AirbyteStreamBase {
         ? streamState?.lastSynced ?? cutoffTimestamp.toISOString()
         : undefined;
 
-    yield* pagerduty.getIncidents(since, this.config.page_size);
+    yield* pagerduty.getIncidents(
+      since,
+      this.config.page_size,
+      this.config.exclude_services
+    );
   }
 
   getUpdatedState(

sources/pagerduty-source/test/index.test.ts

@@ -8,7 +8,7 @@ import fs from 'fs-extra';
 import {VError} from 'verror';
 
 import * as sut from '../src/index';
-import {Pagerduty} from '../src/pagerduty';
+import {Incident, Pagerduty} from '../src/pagerduty';
 
 function readResourceFile(fileName: string): any {
   return JSON.parse(fs.readFileSync(`resources/${fileName}`, 'utf8'));
@@ -147,6 +147,57 @@ describe('index', () => {
     expect(incidents).toStrictEqual(readTestResourceFile('incidents.json'));
   });
 
+  test('streams - incidents, exclude services', async () => {
+    const fnList = jest.fn();
+
+    Pagerduty.instance = jest.fn().mockImplementation(() => {
+      return new Pagerduty(
+        {
+          get: fnList.mockImplementation(async (path: string) => {
+            const incidentsPathMatch = path.match(/^\/incidents/);
+            if (incidentsPathMatch) {
+              const includeServices = decodeURIComponent(path)
+                .split('&')
+                .filter((p) => p.startsWith('service_ids[]='))
+                .map((p) => p.split('=')[1]);
+              return {
+                resource: (
+                  readTestResourceFile('incidents.json') as Incident[]
+                ).filter((i) => includeServices.includes(i.service.id)),
+              };
+            }
+            const servicesPathMatch = path.match(/^\/services/);
+            if (servicesPathMatch) {
+              return {
+                resource: readTestResourceFile('services.json'),
+              };
+            }
+          }),
+        },
+        logger
+      );
+    });
+    const source = new sut.PagerdutySource(logger);
+    const streams = source.streams({
+      token: 'pass',
+      exclude_services: ['Service2'],
+    });
+
+    const incidentsStream = streams[1];
+    const incidentsIter = incidentsStream.readRecords(SyncMode.FULL_REFRESH);
+    const incidents = [];
+    for await (const incident of incidentsIter) {
+      incidents.push(incident);
+    }
+
+    expect(fnList).toHaveBeenCalledTimes(2);
+    expect(incidents).toStrictEqual(
+      (readTestResourceFile('incidents.json') as Incident[]).filter(
+        (i) => i.service.summary !== 'Service2'
+      )
+    );
+  });
+
   test('streams - prioritiesResource, use full_refresh sync mode', async () => {
     const fnPrioritiesResourceList = jest.fn();
 

sources/pagerduty-source/test_files/incidents.json

@@ -240,9 +240,9 @@
     "status": "triggered",
     "incident_key": "4f1a2c8ae595475fa44fb5b82b0330d2",
     "service": {
-      "id": "PAZDKS7",
+      "id": "PAZDKS8",
       "type": "service_reference",
-      "summary": "Service",
+      "summary": "Service2",
       "self": "https://api.pagerduty.com/services/PAZDKS7",
       "html_url": "https://dev-devcube.pagerduty.com/service-directory/PAZDKS7"
     },

sources/pagerduty-source/test_files/services.json

@@ -0,0 +1,16 @@
+[
+  {
+    "id": "PAZDKS7",
+    "type": "service_reference",
+    "summary": "Service1",
+    "self": "https://api.pagerduty.com/services/PAZDKS7",
+    "html_url": "https://dev-devcube.pagerduty.com/service-directory/PAZDKS7"
+  },
+  {
+    "id": "PAZDKS8",
+    "type": "service_reference",
+    "summary": "Service2",
+    "self": "https://api.pagerduty.com/services/PAZDKS8",
+    "html_url": "https://dev-devcube.pagerduty.com/service-directory/PAZDKS8"
+  }
+]