feat(alertmanager): Add exporter config and implementation (#23569)

exporter/alertmanagerexporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

exporter/alertmanagerexporter/README.md

@@ -0,0 +1,49 @@
+# Alertmanager Exporter
+
+Exports Span Events as alerts to [Alertmanager](https://prometheus.io/docs/alerting/latest/alertmanager/) back-end.
+
+Supported pipeline types: traces
+
+## Getting Started
+
+The following settings are required:
+
+- `endpoint` : Alertmanager endpoint to send events
+- `severity` (default info): Default severity for Alerts
+
+
+The following settings are optional:
+
+- `timeout` `sending_queue` and `retry_on_failure` settings as provided by [Exporter Helper](https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/exporterhelper#configuration)
+- [HTTP settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/confighttp/README.md)
+- [TLS and mTLS settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md)
+- `generator_url` is the source of the alerts to be used in Alertmanager's payload and can be set to the URL of the opentelemetry collector if required
+- `severity_attribute`is the spanevent Attribute name which can be used instead of default severity string in Alert payload
+   eg: If severity_attribute is set to "foo" and the SpanEvent has an attribute called foo, foo's attribute value will be used as the severity value for that particular Alert generated from the SpanEvent.
+
+
+
+Example config:
+
+```yaml
+exporters:
+  alertmanager:
+  alertmanager/2:
+    endpoint: "https://a.new.alertmanager.target:9093"
+    severity: "debug"
+    severity_attribute: "foo"
+    tls:
+      cert_file: /var/lib/mycert.pem
+      key_file: /var/lib/key.pem
+    timeout: 10s
+    sending_queue:
+      enabled: true
+      num_consumers: 2
+      queue_size: 10
+    retry_on_failure:
+      enabled: true
+      initial_interval: 10s
+      max_interval: 60s
+      max_elapsed_time: 10m
+    generator_url: "otelcol:55681"
+```

exporter/alertmanagerexporter/alertmanager_exporter.go

@@ -0,0 +1,248 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package alertmanagerexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/alertmanagerexporter"
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+
+	// "go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+type alertmanagerExporter struct {
+	config            *Config
+	client            *http.Client
+	tracesMarshaler   ptrace.Marshaler
+	settings          component.TelemetrySettings
+	endpoint          string
+	generatorUrl      string
+	defaultSeverity   string
+	severityAttribute string
+}
+
+type alertmanagerEvent struct {
+	spanEvent ptrace.SpanEvent
+	traceID   string
+	spanID    string
+	severity  string
+}
+
+func (s *alertmanagerExporter) convertEventSliceToArray(eventslice ptrace.SpanEventSlice, traceID pcommon.TraceID, spanID pcommon.SpanID) []*alertmanagerEvent {
+	if eventslice.Len() > 0 {
+		events := make([]*alertmanagerEvent, eventslice.Len())
+
+		for i := 0; i < eventslice.Len(); i++ {
+			var severity string
+			// severity := pcommon.NewValueStr(s.defaultSeverity)
+			severityAttrValue, ok := eventslice.At(i).Attributes().Get(s.severityAttribute)
+			if ok {
+				severity = severityAttrValue.AsString()
+			} else {
+				severity = s.defaultSeverity
+			}
+			event := alertmanagerEvent{
+				spanEvent: eventslice.At(i),
+				traceID:   traceID.String(),
+				spanID:    spanID.String(),
+				severity:  severity,
+			}
+
+			events[i] = &event
+		}
+		return events
+	}
+	return nil
+}
+
+func (s *alertmanagerExporter) extractEvents(td ptrace.Traces) []*alertmanagerEvent {
+
+	//Stitch parent trace ID and span ID
+	rss := td.ResourceSpans()
+	var events []*alertmanagerEvent = nil
+	if rss.Len() == 0 {
+		return nil
+	}
+
+	for i := 0; i < rss.Len(); i++ {
+		resource := rss.At(i).Resource()
+		ilss := rss.At(i).ScopeSpans()
+
+		if resource.Attributes().Len() == 0 && ilss.Len() == 0 {
+			return nil
+		}
+
+		for j := 0; j < ilss.Len(); j++ {
+			spans := ilss.At(j).Spans()
+			for k := 0; k < spans.Len(); k++ {
+				traceID := pcommon.TraceID(spans.At(k).TraceID())
+				spanID := pcommon.SpanID(spans.At(k).SpanID())
+				events = append(events, s.convertEventSliceToArray(spans.At(k).Events(), traceID, spanID)...)
+			}
+		}
+	}
+	return events
+}
+
+func createAnnotations(event *alertmanagerEvent) model.LabelSet {
+	LabelMap := make(model.LabelSet, event.spanEvent.Attributes().Len()+1)
+	event.spanEvent.Attributes().Range(func(key string, attr pcommon.Value) bool {
+		LabelMap[model.LabelName(key)] = model.LabelValue(attr.AsString())
+		return true
+	})
+	LabelMap["TraceID"] = model.LabelValue(event.traceID)
+	LabelMap["SpanID"] = model.LabelValue(event.spanID)
+	return LabelMap
+}
+
+func (s *alertmanagerExporter) convertEventstoAlertPayload(events []*alertmanagerEvent) []model.Alert {
+
+	var payload []model.Alert
+	for _, event := range events {
+		annotations := createAnnotations(event)
+
+		alert := model.Alert{
+			StartsAt:     time.Now(),
+			Labels:       model.LabelSet{"severity": model.LabelValue(event.severity), "event_name": model.LabelValue(event.spanEvent.Name())},
+			Annotations:  annotations,
+			GeneratorURL: s.generatorUrl,
+		}
+
+		payload = append(payload, alert)
+	}
+	return payload
+}
+
+func (s *alertmanagerExporter) postAlert(ctx context.Context, payload []model.Alert) error {
+	msg, _ := json.Marshal(payload)
+
+	req, err := http.NewRequestWithContext(ctx, "POST", s.endpoint, bytes.NewBuffer(msg))
+	if err != nil {
+		s.settings.Logger.Debug("error creating HTTP request", zap.Error(err))
+		return fmt.Errorf("error creating HTTP request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := s.client.Do(req)
+	if err != nil {
+		s.settings.Logger.Debug("error sending HTTP request", zap.Error(err))
+		return fmt.Errorf("error sending HTTP request: %w", err)
+	}
+
+	defer func() {
+		if closeErr := resp.Body.Close(); closeErr != nil {
+			s.settings.Logger.Warn("Failed to close response body", zap.Error(closeErr))
+		}
+	}()
+
+	_, err = ioutil.ReadAll(resp.Body)
+	if err != nil {
+		s.settings.Logger.Debug("failed to read response body", zap.Error(err))
+		return fmt.Errorf("failed to read response body %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		s.settings.Logger.Debug("post request to Alertmanager failed", zap.Error(err))
+		return fmt.Errorf("request POST %s failed - %q", req.URL.String(), resp.Status)
+	}
+	return nil
+}
+
+func (s *alertmanagerExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
+
+	s.settings.Logger.Debug("TracesExporter", zap.Int("#spans", td.SpanCount()))
+
+	events := s.extractEvents(td)
+
+	if len(events) == 0 {
+		return nil
+	}
+
+	alert := s.convertEventstoAlertPayload(events)
+	err := s.postAlert(ctx, alert)
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *alertmanagerExporter) start(_ context.Context, host component.Host) error {
+
+	client, err := s.config.HTTPClientSettings.ToClient(host, s.settings)
+	if err != nil {
+		s.settings.Logger.Error("failed to create HTTP Client", zap.Error(err))
+		return fmt.Errorf("failed to create HTTP Client: %w", err)
+	}
+	s.client = client
+	return nil
+}
+
+func (s *alertmanagerExporter) shutdown(context.Context) error {
+	return nil
+}
+
+func newAlertManagerExporter(cfg *Config, set component.TelemetrySettings) *alertmanagerExporter {
+
+	url := cfg.GeneratorURL
+
+	if len(url) == 0 {
+		url = "http://example.com/alert"
+	}
+
+	severity := cfg.DefaultSeverity
+
+	if len(severity) == 0 {
+		severity = "info"
+	}
+
+	return &alertmanagerExporter{
+		config:            cfg,
+		settings:          set,
+		tracesMarshaler:   &ptrace.JSONMarshaler{},
+		endpoint:          fmt.Sprintf("%s/api/v1/alerts", cfg.HTTPClientSettings.Endpoint),
+		generatorUrl:      url,
+		defaultSeverity:   severity,
+		severityAttribute: cfg.SeverityAttribute,
+	}
+}
+
+func newTracesExporter(ctx context.Context, cfg component.Config, set exporter.CreateSettings) (exporter.Traces, error) {
+	config := cfg.(*Config)
+
+	if len(config.HTTPClientSettings.Endpoint) == 0 {
+		return nil, errors.New("endpoint is not set")
+	}
+
+	s := newAlertManagerExporter(config, set.TelemetrySettings)
+
+	return exporterhelper.NewTracesExporter(
+		ctx,
+		set,
+		cfg,
+		s.pushTraces,
+		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
+		// Disable Timeout/RetryOnFailure and SendingQueue
+		exporterhelper.WithStart(s.start),
+		exporterhelper.WithTimeout(config.TimeoutSettings),
+		exporterhelper.WithRetry(config.RetrySettings),
+		exporterhelper.WithQueue(config.QueueSettings),
+		exporterhelper.WithShutdown(s.shutdown),
+	)
+}