Add Pagerduty source and destination

destinations/faros-destination/resources/spec.json

@@ -97,6 +97,34 @@
                 "default": false
               }
             }
+          },
+          "pagerduty": {
+            "title": "PagerDuty",
+            "description": "Configuration options that apply to records generated by the PagerDuty Source.",
+            "type": "object",
+            "properties": {
+              "application_mapping": {
+                "type": "object",
+                "title": "Application Mapping",
+                "description": "JSON map of PagerDuty service(s) name, to compute platform specific app name and platform name. Used to reference compute_Application object, from an ims_Incident object.",
+                "default": {},
+                "examples": [
+                  {
+                    "Aion": {
+                      "name": "aion",
+                      "platform": "ECS"
+                    }
+                  }
+                ]
+              },
+              "default_severity": {
+                "type": "string",
+                "title": "Default Severity",
+                "description": "A default severity category if not present",
+                "pattern": "^(Sev[1-5])?(Custom)?$",
+                "examples": ["Sev1", "Sev2", "Sev3", "Sev4", "Sev5", "Custom"]
+              }
+            }
           }
         }
       }

destinations/faros-destination/src/converters/pagerduty/common.ts

@@ -0,0 +1,51 @@
+import {AirbyteRecord} from 'faros-airbyte-cdk';
+
+import {Converter, StreamContext} from '../converter';
+
+export interface PagerdutyObject {
+  readonly id: string;
+  readonly type: string; // object type of the form <name>_reference
+  readonly summary: string; // human readable summary
+  readonly self: string; // API discrete resource url
+  readonly html_url: string; // PagerDuty web url
+}
+
+export enum IncidentSeverityCategory {
+  Sev1 = 'Sev1',
+  Sev2 = 'Sev2',
+  Sev3 = 'Sev3',
+  Sev4 = 'Sev4',
+  Sev5 = 'Sev5',
+  Custom = 'Custom',
+}
+
+type ApplicationMapping = Record<string, {name: string; platform?: string}>;
+
+interface PagerdutyConfig {
+  application_mapping?: ApplicationMapping;
+  default_severity?: IncidentSeverityCategory;
+}
+
+/** PagerDuty converter base */
+export abstract class PagerdutyConverter extends Converter {
+  /** Almost every Pagerduty record have id property. Function will be
+   * override if record doesn't have id property.
+   */
+  id(record: AirbyteRecord): any {
+    return record?.record?.data?.id;
+  }
+
+  protected pagerdutyConfig(ctx: StreamContext): PagerdutyConfig {
+    return ctx.config.source_specific_configs?.pagerduty ?? {};
+  }
+
+  protected applicationMapping(ctx: StreamContext): ApplicationMapping {
+    return this.pagerdutyConfig(ctx).application_mapping ?? {};
+  }
+
+  protected defaultSeverity(
+    ctx: StreamContext
+  ): IncidentSeverityCategory | null {
+    return this.pagerdutyConfig(ctx).default_severity ?? null;
+  }
+}

destinations/faros-destination/src/converters/pagerduty/incident_log_entries.ts

@@ -0,0 +1,62 @@
+import {AirbyteRecord} from 'faros-airbyte-cdk';
+import {Utils} from 'faros-feeds-sdk';
+
+import {DestinationModel, DestinationRecord, StreamContext} from '../converter';
+import {PagerdutyConverter} from './common';
+
+interface IncidentEventType {
+  category: IncidentEventTypeCategory;
+  detail: string;
+}
+
+enum IncidentEventTypeCategory {
+  Created = 'Created',
+  Acknowledged = 'Acknowledged',
+  Resolved = 'Resolved',
+  Custom = 'Custom',
+}
+
+export class PagerdutyIncidentLogEntries extends PagerdutyConverter {
+  readonly destinationModels: ReadonlyArray<DestinationModel> = ['ims_IncidentEvent'];
+
+  convert(
+    record: AirbyteRecord,
+    ctx: StreamContext
+  ): ReadonlyArray<DestinationRecord> {
+    const source = this.streamName.source;
+    const event = record.record.data;
+
+    return [
+      {
+        model: 'ims_IncidentEvent',
+        record: {
+          uid: event.id,
+          type: this.eventType(event.type),
+          createdAt: Utils.toDate(event.created_at),
+          detail: event.summary,
+          incident: {uid: event.incident.id, source},
+        },
+      },
+    ];
+  }
+
+  private eventType(logEntryType: string): IncidentEventType {
+    const typeRef = logEntryType.split('_')[0]; // i.e. "resolve" of "resolve_log_entry"
+    let eventTypeCategory;
+    switch (typeRef.toLowerCase()) {
+      case 'trigger':
+        eventTypeCategory = IncidentEventTypeCategory.Created;
+        break;
+      case 'acknowledge':
+        eventTypeCategory = IncidentEventTypeCategory.Acknowledged;
+        break;
+      case 'resolve':
+        eventTypeCategory = IncidentEventTypeCategory.Resolved;
+        break;
+      default:
+        eventTypeCategory = IncidentEventTypeCategory.Custom;
+        break;
+    }
+    return {category: eventTypeCategory, detail: logEntryType};
+  }
+}

destinations/faros-destination/src/converters/pagerduty/incidents.ts

@@ -0,0 +1,152 @@
+import {AirbyteLogger, AirbyteRecord} from 'faros-airbyte-cdk';
+import {Utils} from 'faros-feeds-sdk';
+
+import {DestinationModel, DestinationRecord, StreamContext} from '../converter';
+import {PagerdutyConverter, PagerdutyObject} from './common';
+
+enum IncidentStatusCategory {
+  Created = 'Created',
+  Identified = 'Identified',
+  Investigating = 'Investigating',
+  Resolved = 'Resolved',
+  Custom = 'Custom',
+}
+
+type IncidentUrgency = 'high' | 'low'; //PagerDuty only has these two priorities
+type IncidentState = 'triggered' | 'acknowledged' | 'resolved';
+
+interface Acknowledgement {
+  at: string; //date-time
+  acknowledger: PagerdutyObject;
+}
+
+export class PagerdutyIncidents extends PagerdutyConverter {
+  private readonly logger = new AirbyteLogger();
+
+  readonly destinationModels: ReadonlyArray<DestinationModel> = [
+    'compute_Application',
+    'ims_Incident',
+    'ims_IncidentApplicationImpact',
+    'ims_IncidentAssignment',
+  ];
+
+  convert(
+    record: AirbyteRecord,
+    ctx: StreamContext
+  ): ReadonlyArray<DestinationRecord> {
+    const source = this.streamName.source;
+    const incident = record.record.data;
+    const res: DestinationRecord[] = [];
+    const incidentRef = {uid: incident.id, source};
+    const lastUpdated = Utils.toDate(incident.last_status_change_at);
+
+    let acknowledgedAt, resolvedAt;
+    if (incident.status === 'acknowledged') {
+      if (!incident.acknowledgements || !incident.acknowledgements.length) {
+        this.logger.warn(
+          `Incident ${incident.id} acknowledged, but acknowledger info missing`
+        );
+      } else {
+        // find first acknowledgement
+        acknowledgedAt = Utils.toDate(
+          incident.acknowledgements
+            .map((ack: Acknowledgement) => ack.at)
+            .sort()[0]
+        );
+      }
+    } else if (incident.status === 'resolved') {
+      resolvedAt = lastUpdated;
+    }
+
+    res.push({
+      // We are explicitly passing __Upsert command here with at := 0,
+      // to allow updating Incident severity from prioritiesResource stream
+      // in the same revision
+      model: 'ims_Incident__Upsert',
+      record: {
+        at: 0,
+        data: {
+          ...incidentRef,
+          title: incident.title,
+          description: incident.description,
+          url: incident.self,
+          createdAt: Utils.toDate(incident.created_at),
+          updatedAt: resolvedAt,
+          acknowledgedAt: acknowledgedAt,
+          resolvedAt: resolvedAt,
+          priority: this.incidentPriority(incident.urgency),
+          status: this.incidentState(incident.status),
+        },
+      },
+    });
+
+    for (const assignment of incident.assignments) {
+      const assignee = {uid: assignment.assignee.id, source};
+      res.push({
+        model: 'ims_IncidentAssignment',
+        record: {
+          incident: incidentRef,
+          assignee,
+        },
+      });
+    }
+
+    const applicationMapping = this.applicationMapping(ctx);
+    let application = {
+      name: incident.service.summary,
+      platform: '',
+    };
+    // if we have an app mapping specified
+    if (
+      incident.service.summary in applicationMapping &&
+      applicationMapping[incident.service.summary].name
+    ) {
+      const mappedApp = applicationMapping[incident.service.summary];
+      application = {
+        name: mappedApp.name,
+        platform: mappedApp.platform ?? application.platform,
+      };
+    }
+    res.push({model: 'compute_Application', record: application});
+
+    res.push({
+      model: 'ims_IncidentApplicationImpact',
+      record: {
+        incident: incidentRef,
+        application,
+      },
+    });
+
+    return res;
+  }
+
+  private incidentPriority(
+    incidentUrgency: IncidentUrgency
+  ): Record<string, string> {
+    const detail = incidentUrgency;
+    switch (incidentUrgency.toLowerCase()) {
+      case 'high':
+        return {category: 'High', detail};
+      case 'low':
+        return {category: 'Low', detail};
+      default:
+        return {category: 'Custom', detail};
+    }
+  }
+
+  private incidentState(incidentStatus: IncidentState): {
+    category: string;
+    detail: string;
+  } {
+    const detail = incidentStatus;
+    switch (incidentStatus) {
+      case 'resolved':
+        return {category: IncidentStatusCategory.Resolved, detail};
+      case 'acknowledged':
+        return {category: IncidentStatusCategory.Investigating, detail};
+      case 'triggered':
+      default:
+        return {category: IncidentStatusCategory.Created, detail};
+    }
+  }
+}