Add support for PagerDuty

Signed-off-by: Martin Kemp <me@martinke.mp>

api/v1beta2/provider_types.go

@@ -48,12 +48,13 @@ const (
 	Matrix                 string = "matrix"
 	OpsgenieProvider       string = "opsgenie"
 	AlertManagerProvider   string = "alertmanager"
+	PagerDutyProvider      string = "pagerduty"
 )
 
 // ProviderSpec defines the desired state of the Provider.
 type ProviderSpec struct {
 	// Type specifies which Provider implementation to use.
-	// +kubebuilder:validation:Enum=slack;discord;msteams;rocket;generic;generic-hmac;github;gitlab;gitea;bitbucket;azuredevops;googlechat;googlepubsub;webex;sentry;azureeventhub;telegram;lark;matrix;opsgenie;alertmanager;grafana;githubdispatch;
+	// +kubebuilder:validation:Enum=slack;discord;msteams;rocket;generic;generic-hmac;github;gitlab;gitea;bitbucket;azuredevops;googlechat;googlepubsub;webex;sentry;azureeventhub;telegram;lark;matrix;opsgenie;alertmanager;grafana;githubdispatch;pagerduty
 	// +required
 	Type string `json:"type"`
 

config/crd/bases/notification.toolkit.fluxcd.io_providers.yaml

@@ -302,6 +302,7 @@ spec:
                 - alertmanager
                 - grafana
                 - githubdispatch
+                - pagerduty
                 type: string
               username:
                 description: Username specifies the name under which events are posted.

docs/spec/v1beta2/providers.md

@@ -118,6 +118,7 @@ The supported alerting providers are:
 | [Matrix](#matrix)                                       | `matrix`         |
 | [Microsoft Teams](#microsoft-teams)                     | `msteams`        |
 | [Opsgenie](#opsgenie)                                   | `opsgenie`       |
+| [PagerDuty](#pagerduty)                                 | `pagerduty`      |
 | [Prometheus Alertmanager](#prometheus-alertmanager)     | `alertmanager`   |
 | [Rocket](#rocket)                                       | `rocket`         |
 | [Sentry](#sentry)                                       | `sentry`         |
@@ -772,6 +773,64 @@ stringData:
     token: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
 ```
 
+##### PagerDuty
+
+When `.spec.type` is set to `pagerduty`, the controller will send a payload for
+an [Event](events.md#event-structure) to the provided PagerDuty [Address](#address).
+
+The Event will be formatted into an [Event API v2](https://developer.pagerduty.com/api-reference/368ae3d938c9e-send-an-event-to-pager-duty) payload,
+triggering or resolving an incident depending on the event's `Severity`.
+
+The provider will also send [Change Events](https://developer.pagerduty.com/api-reference/95db350959c37-send-change-events-to-the-pager-duty-events-api)
+for `info` level `Severity`, which will be displayed in the PagerDuty service's timeline to track changes.
+
+This Provider type supports the configuration of a [proxy URL](#https-proxy)
+and [TLS certificates](#tls-certificates).
+
+The [Channel](#channel) is used to set the routing key to send the event to the appropriate integration.
+
+###### PagerDuty example
+
+To configure a Provider for Pagerduty, create a `pagerduty` Provider,
+set `address` to the integration URL and `channel` set to
+the integration key (also known as a routing key) for your [service](https://support.pagerduty.com/docs/services-and-integrations#create-a-generic-events-api-integration)
+or [event orchestration](https://support.pagerduty.com/docs/event-orchestration).
+
+When adding an integration for a service on PagerDuty, it is recommended to use `Events API v2` integration.
+
+**Note**: PagerDuty does not support Change Events when sent to global integrations, such as event orchestration.
+
+```yaml
+---
+apiVersion: notification.toolkit.fluxcd.io/v1beta2
+kind: Provider
+metadata:
+  name: pagerduty
+  namespace: default
+spec:
+  type: pagerduty
+  address: https://events.pagerduty.com
+  channel: <integrationKey>
+```
+If you are sending to a service integration, it is recommended to set your Alert to filter to
+only those sources you want to trigger an incident for that service. For example:
+
+```yaml
+---
+apiVersion: notification.toolkit.fluxcd.io/v1beta2
+kind: Alert
+metadata:
+  name: my-service-pagerduty
+  namespace: default
+spec:
+  providerRef:
+    name: pagerduty
+  eventSources:
+    - kind: HelmRelease
+      name: my-service
+      namespace: default
+```
+
 ##### Prometheus Alertmanager
 
 When `.spec.type` is set to `alertmanager`, the controller will send a payload for

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230106234847-43070de90fa1
 	github.com/Azure/azure-amqp-common-go/v4 v4.2.0
 	github.com/Azure/azure-event-hubs-go/v3 v3.6.0
+	github.com/PagerDuty/go-pagerduty v1.7.0
 	github.com/containrrr/shoutrrr v0.7.1
 	github.com/fluxcd/notification-controller/api v1.0.0-rc.4
 	github.com/fluxcd/pkg/apis/event v0.5.1

go.sum

@@ -237,6 +237,8 @@ github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3
 github.com/MakeNowJust/heredoc v1.0.0 h1:cXCdzVdstXyiTqTvfqk9SDHpKNjxuom+DOlyEeQ4pzQ=
 github.com/MakeNowJust/heredoc v1.0.0/go.mod h1:mG5amYoWBHf8vpLOuehzbGGw0EHxpZZ6lCpQ4fNJ8LE=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
+github.com/PagerDuty/go-pagerduty v1.7.0 h1:S1NcMKECxT5hJwV4VT+QzeSsSiv4oWl1s2821dUqG/8=
+github.com/PagerDuty/go-pagerduty v1.7.0/go.mod h1:PuFyJKRz1liIAH4h5KVXVD18Obpp1ZXRdxHvmGXooro=
 github.com/ProtonMail/go-crypto v0.0.0-20230619160724-3fbb1f12458c h1:figwFwYep1Qnl64Y+Rc8tyQWE0xvYAN+5EX+rD40pTU=
 github.com/ProtonMail/go-crypto v0.0.0-20230619160724-3fbb1f12458c/go.mod h1:EjAoLdwvbIOoOQr3ihjnSoLZRtE8azugULFRteWMNc0=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

internal/notifier/factory.go

@@ -111,6 +111,8 @@ func (f Factory) Notifier(provider string) (Interface, error) {
 		n, err = NewAlertmanager(f.URL, f.ProxyURL, f.CertPool)
 	case apiv1.GrafanaProvider:
 		n, err = NewGrafana(f.URL, f.ProxyURL, f.Token, f.CertPool, f.Username, f.Password)
+	case apiv1.PagerDutyProvider:
+		n, err = NewPagerDuty(f.URL, f.ProxyURL, f.CertPool, f.Channel)
 	default:
 		err = fmt.Errorf("provider %s not supported", provider)
 	}

internal/notifier/pagerduty.go

@@ -0,0 +1,126 @@
+/*
+Copyright 2023 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package notifier
+
+import (
+	"context"
+	"crypto/x509"
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/PagerDuty/go-pagerduty"
+
+	eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
+	"github.com/fluxcd/pkg/apis/meta"
+)
+
+type PagerDuty struct {
+	Endpoint   string
+	RoutingKey string
+	ProxyURL   string
+	CertPool   *x509.CertPool
+}
+
+func NewPagerDuty(endpoint string, proxyURL string, certPool *x509.CertPool, routingKey string) (*PagerDuty, error) {
+	URL, err := url.ParseRequestURI(endpoint)
+	if err != nil {
+		return nil, fmt.Errorf("invalid PagerDuty endpoint URL %q: '%w'", endpoint, err)
+	}
+	return &PagerDuty{
+		Endpoint:   URL.Scheme + "://" + URL.Host,
+		RoutingKey: routingKey,
+		ProxyURL:   proxyURL,
+		CertPool:   certPool,
+	}, nil
+}
+
+func (p *PagerDuty) Post(ctx context.Context, event eventv1.Event) error {
+	// skip commit status updates and progressing events (we want success or failure)
+	if event.HasMetadata(eventv1.MetaCommitStatusKey, eventv1.MetaCommitStatusUpdateValue) || event.HasReason(meta.ProgressingReason) {
+		return nil
+	}
+	e := toPagerDutyV2Event(event, p.RoutingKey)
+	err := postMessage(ctx, p.Endpoint+"/v2/enqueue", p.ProxyURL, p.CertPool, e)
+	if err != nil {
+		return fmt.Errorf("failed sending event: %w", err)
+	}
+	// Send a change event for info events
+	if event.Severity == eventv1.EventSeverityInfo {
+		ce := toPagerDutyChangeEvent(event, p.RoutingKey)
+		err = postMessage(ctx, p.Endpoint+"/v2/change/enqueue", p.ProxyURL, p.CertPool, ce)
+		if err != nil {
+			return fmt.Errorf("failed sending change event: %w", err)
+		}
+	}
+	return nil
+}
+
+func toPagerDutyV2Event(event eventv1.Event, routingKey string) pagerduty.V2Event {
+	name, desc := formatNameAndDescription(event)
+	// Send resolve just in case an existing incident is open
+	e := pagerduty.V2Event{
+		RoutingKey: routingKey,
+		Action:     "resolve",
+		DedupKey:   string(event.InvolvedObject.UID),
+	}
+	// Trigger an incident for errors
+	if event.Severity == eventv1.EventSeverityError {
+		e.Action = "trigger"
+		e.Payload = &pagerduty.V2Payload{
+			Summary:   desc + ": " + name,
+			Source:    "Flux " + event.ReportingController,
+			Severity:  toPagerDutySeverity(event.Severity),
+			Timestamp: event.Timestamp.Format(time.RFC3339),
+			Component: event.InvolvedObject.Name,
+			Group:     event.InvolvedObject.Kind,
+			Details: map[string]interface{}{
+				"message":  event.Message,
+				"metadata": event.Metadata,
+			},
+		}
+	}
+	return e
+}
+
+func toPagerDutyChangeEvent(event eventv1.Event, routingKey string) pagerduty.ChangeEvent {
+	name, desc := formatNameAndDescription(event)
+	ce := pagerduty.ChangeEvent{
+		RoutingKey: routingKey,
+		Payload: pagerduty.ChangeEventPayload{
+			Summary:   desc + ": " + name,
+			Source:    "Flux " + event.ReportingController,
+			Timestamp: event.Timestamp.Format(time.RFC3339),
+			CustomDetails: map[string]interface{}{
+				"message":  event.Message,
+				"metadata": event.Metadata,
+			},
+		},
+	}
+	return ce
+}
+
+func toPagerDutySeverity(severity string) string {
+	switch severity {
+	case eventv1.EventSeverityError:
+	case eventv1.EventSeverityInfo:
+		return severity
+	case eventv1.EventSeverityTrace:
+		return "info"
+	}
+	return "error"
+}

internal/notifier/pagerduty_fuzz_test.go

@@ -0,0 +1,56 @@
+package notifier
+
+import (
+	"context"
+	"crypto/x509"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	fuzz "github.com/AdaLogics/go-fuzz-headers"
+	eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
+)
+
+func Fuzz_PagerDuty(f *testing.F) {
+	f.Add("token", "", "error", "", []byte{}, []byte{})
+	f.Add("token", "", "info", "", []byte{}, []byte{})
+
+	f.Fuzz(func(t *testing.T,
+		routingKey, commitStatus, severity, message string, seed, response []byte) {
+		mux := http.NewServeMux()
+		mux.HandleFunc("/v2/enqueue", func(w http.ResponseWriter, r *http.Request) {
+			w.Write(response)
+			io.Copy(io.Discard, r.Body)
+			r.Body.Close()
+		})
+		mux.HandleFunc("/v2/change/enqueue", func(w http.ResponseWriter, r *http.Request) {
+			w.Write(response)
+			io.Copy(io.Discard, r.Body)
+			r.Body.Close()
+		})
+		ts := httptest.NewServer(mux)
+		defer ts.Close()
+
+		var cert x509.CertPool
+		_ = fuzz.NewConsumer(seed).GenerateStruct(&cert)
+
+		pd, err := NewPagerDuty(ts.URL, "", &cert, routingKey)
+		if err != nil {
+			return
+		}
+
+		event := eventv1.Event{}
+		_ = fuzz.NewConsumer(seed).GenerateStruct(&event)
+
+		if event.Metadata == nil {
+			event.Metadata = map[string]string{}
+		}
+
+		event.Metadata["commit_status"] = commitStatus
+		event.Message = message
+		event.Severity = severity
+
+		_ = pd.Post(context.TODO(), event)
+	})
+}

internal/notifier/pagerduty_test.go

@@ -0,0 +1,228 @@
+package notifier
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/PagerDuty/go-pagerduty"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
+	"github.com/fluxcd/pkg/apis/meta"
+)
+
+const (
+	routingKey                = "notARealRoutingKey"
+	pagerdutyEUv2EventsAPIURL = "https://events.eu.pagerduty.com"
+)
+
+func TestNewPagerDuty(t *testing.T) {
+	t.Run("US endpoint", func(t *testing.T) {
+		p, err := NewPagerDuty("https://events.pagerduty.com/v2/enqueue", "", nil, routingKey)
+		assert.NoError(t, err)
+		assert.Equal(t, routingKey, p.RoutingKey)
+		assert.NotEqual(t, pagerdutyEUv2EventsAPIURL, p.Endpoint)
+	})
+	t.Run("EU endpoint", func(t *testing.T) {
+		p, err := NewPagerDuty("https://events.eu.pagerduty.com/v2/enqueue", "", nil, routingKey)
+		assert.NoError(t, err)
+		assert.Equal(t, routingKey, p.RoutingKey)
+		assert.Equal(t, pagerdutyEUv2EventsAPIURL, p.Endpoint)
+	})
+	t.Run("invalid URL", func(t *testing.T) {
+		_, err := NewPagerDuty("not a url", "", nil, routingKey)
+		assert.Errorf(t, err, "invalid PagerDuty endpoint URL not a url: 'parse \"https://not a url/\": invalid character \" \" in host name'")
+	})
+}
+
+func TestPagerDutyPost(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/v2/enqueue", func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		var payload pagerduty.V2Event
+		err = json.Unmarshal(b, &payload)
+		require.NoError(t, err)
+	})
+	mux.HandleFunc("/v2/change/enqueue", func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		var payload pagerduty.ChangeEvent
+		err = json.Unmarshal(b, &payload)
+		require.NoError(t, err)
+	})
+	ts := httptest.NewServer(mux)
+	defer ts.Close()
+
+	pd, err := NewPagerDuty(ts.URL, "", nil, "token")
+	require.NoError(t, err)
+
+	err = pd.Post(context.TODO(), testEvent())
+	require.NoError(t, err)
+}
+
+func TestToPagerDutyV2Event(t *testing.T) {
+	// Construct test event
+	tests := []struct {
+		name string
+		e    eventv1.Event
+		want pagerduty.V2Event
+	}{
+		{
+			name: "basic",
+			e: eventv1.Event{
+				InvolvedObject: corev1.ObjectReference{
+					Kind:      "GitRepository",
+					Namespace: "flux-system",
+					Name:      "test-app",
+					UID:       "1234",
+				},
+				Severity:  "info",
+				Timestamp: metav1.Date(2020, 01, 01, 0, 0, 0, 0, time.UTC),
+				Message:   "message",
+				Reason:    meta.SucceededReason,
+				Metadata: map[string]string{
+					"key1": "val1",
+					"key2": "val2",
+				},
+				ReportingController: "source-controller",
+			},
+			want: pagerduty.V2Event{
+				RoutingKey: routingKey,
+				Action:     "resolve",
+				DedupKey:   "1234",
+			},
+		},
+		{
+			name: "error",
+			e: eventv1.Event{
+				InvolvedObject: corev1.ObjectReference{
+					Kind:      "GitRepository",
+					Namespace: "flux-system",
+					Name:      "test-app",
+					UID:       "1234",
+				},
+				Severity:  "error",
+				Timestamp: metav1.Date(2020, 01, 01, 0, 0, 0, 0, time.UTC),
+				Message:   "message",
+				Reason:    meta.FailedReason,
+				Metadata: map[string]string{
+					"key1": "val1",
+					"key2": "val2",
+				},
+				ReportingController: "source-controller",
+			},
+			want: pagerduty.V2Event{
+				RoutingKey: routingKey,
+				Action:     "trigger",
+				DedupKey:   "1234",
+				Payload: &pagerduty.V2Payload{
+					Summary:   "failed: gitrepository/test-app",
+					Severity:  "error",
+					Source:    "Flux source-controller",
+					Timestamp: "2020-01-01T00:00:00Z",
+					Component: "test-app",
+					Group:     "GitRepository",
+					Details: map[string]interface{}{
+						"message": "message",
+						"metadata": map[string]string{
+							"key1": "val1",
+							"key2": "val2",
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := toPagerDutyV2Event(tt.e, routingKey)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Logf("got Payload: %+v", got.Payload)
+				t.Errorf("toPagerDutyV2Event() = %+v, want %+v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestToPagerDutyChangeEvent(t *testing.T) {
+	e := eventv1.Event{
+		InvolvedObject: corev1.ObjectReference{
+			Kind:      "GitRepository",
+			Namespace: "flux-system",
+			Name:      "test-app",
+			UID:       "1234",
+		},
+		Severity:  "info",
+		Timestamp: metav1.Date(2020, 01, 01, 0, 0, 0, 0, time.UTC),
+		Message:   "message",
+		Reason:    meta.SucceededReason,
+		Metadata: map[string]string{
+			"key1": "val1",
+			"key2": "val2",
+		},
+		ReportingController: "source-controller",
+	}
+	want := pagerduty.ChangeEvent{
+		RoutingKey: routingKey,
+		Payload: pagerduty.ChangeEventPayload{
+			Summary:   "succeeded: gitrepository/test-app",
+			Source:    "Flux source-controller",
+			Timestamp: "2020-01-01T00:00:00Z",
+			CustomDetails: map[string]interface{}{
+				"message": "message",
+				"metadata": map[string]string{
+					"key1": "val1",
+					"key2": "val2",
+				},
+			},
+		},
+	}
+	got := toPagerDutyChangeEvent(e, routingKey)
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("toPagerDutyChangeEvent() = %q, want %q", got, want)
+	}
+}
+
+func TestToPagerDutySeverity(t *testing.T) {
+	tests := []struct {
+		name     string
+		severity string
+		want     string
+	}{
+		{
+			name:     "info",
+			severity: eventv1.EventSeverityInfo,
+			want:     "info",
+		},
+		{
+			name:     "error",
+			severity: eventv1.EventSeverityError,
+			want:     "error",
+		},
+		{
+			name:     "trace",
+			severity: eventv1.EventSeverityTrace,
+			want:     "info",
+		},
+		{
+			name:     "invalid",
+			severity: "invalid",
+			want:     "error",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, toPagerDutySeverity(tt.severity))
+		})
+	}
+}