[receiver/awss3receiver]: Add ingest progress notifications via OpAMP

.chloggen/awss3receiver_notifications.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'new_component'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awss3receiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: 'Add support for monitoring the progress of ingesting data from an S3 bucket via OpAMP custom messages.'
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30750]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/awss3receiver/README.md

@@ -29,6 +29,8 @@ The following exporter configuration parameters are supported.
 | `endpoint`              | overrides the endpoint used by the exporter instead of constructing it from `region` and `s3_bucket`                                       |             | Optional |
 | `endpoint_partition_id` | partition id to use if `endpoint` is specified.                                                                                            | "aws"       | Optional |
 | `s3_force_path_style`   | [set this to `true` to force the request to use path-style addressing](http://docs.aws.amazon.com/AmazonS3/latest/dev/VirtualHosting.html) | false       | Optional |
+| `notifications:`        |                                                                                                                                            |             |          |
+| `opamp`                 | Name of the OpAMP Extension to use to send notifications of ingest progress.                                                               |             |          |
 
 ### Time format for `starttime` and `endtime`
 The `starttime` and `endtime` fields are used to specify the time range for which to retrieve data. 
@@ -46,4 +48,23 @@ receivers:
         s3_bucket: "mybucket"
         s3_prefix: "trace"
         s3_partition: "minute"
-```
+```
+
+## Notifications
+The receiver can send notifications of ingest progress to an OpAmp server using the custom message capability of 
+"org.opentelemetry.collector.receiver.awss3" and message type "TimeBasedIngestStatus".
+The format of the notifications is a JSON object with the following fields:
+
+| Field             | Description                                                                     |
+|:------------------|:--------------------------------------------------------------------------------|
+| `telemetry_type`  | The type of telemetry being ingested. One of "traces", "metrics", or "logs".    |
+| `ingest_status`   | The status of the data ingestion. One of "ingesting", "failed", or "completed". |
+| `start_time`      | The time to start retrieving data in RFC3339 format.                            |
+| `stop_time`       | The time to stop retrieving data in RFC3339 format.                             |
+| `ingest_time`     | The time of the data currently being ingested in RFC3339 format.                |
+| `failure_message` | Error message if `ingest_status` is "failed".                                   |
+
+The "ingesting" status is sent at the beginning of the ingest process before data has been retrieved for the specified time.
+If during the processing of the data an error occurs a status message with `ingest_status` set to "failed" status with 
+the time of the data being ingested when the failure occurred.
+If the ingest process completes successfully a status message with `ingest_status` set to "completed" is sent.

receiver/awss3receiver/config.go

@@ -26,11 +26,16 @@ type S3DownloaderConfig struct {
 	S3ForcePathStyle    bool   `mapstructure:"s3_force_path_style"`
 }
 
+type Notifications struct {
+	OpAMP *component.ID `mapstructure:"opamp"`
+}
+
 // Config defines the configuration for the file receiver.
 type Config struct {
-	S3Downloader S3DownloaderConfig `mapstructure:"s3downloader"`
-	StartTime    string             `mapstructure:"starttime"`
-	EndTime      string             `mapstructure:"endtime"`
+	S3Downloader  S3DownloaderConfig `mapstructure:"s3downloader"`
+	StartTime     string             `mapstructure:"starttime"`
+	EndTime       string             `mapstructure:"endtime"`
+	Notifications Notifications      `mapstructure:"notifications"`
 }
 
 const (

receiver/awss3receiver/examples/notifications/Makefile

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

receiver/awss3receiver/examples/notifications/go.mod

@@ -0,0 +1,11 @@
+module opampServer
+
+go 1.21.0
+
+require github.com/open-telemetry/opamp-go v0.14.0
+
+require (
+	github.com/gorilla/websocket v1.5.1 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
+)

receiver/awss3receiver/examples/notifications/main.go

@@ -0,0 +1,134 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"time"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/open-telemetry/opamp-go/server"
+	"github.com/open-telemetry/opamp-go/server/types"
+)
+
+const (
+	openTelemetryCollectorReceiverAWSS3 = "org.opentelemetry.collector.receiver.awss3"
+)
+
+type TimeBasedIngestStatus struct {
+	TelemetryType  string    `json:"telemetry_type"`
+	IngestStatus   string    `json:"ingest_status"`
+	StartTime      time.Time `json:"start_time"`
+	EndTime        time.Time `json:"end_time"`
+	IngestTime     time.Time `json:"ingest_time"`
+	FailureMessage string    `json:"failure_message,omitempty"`
+}
+
+type ProgressServer struct {
+	server server.OpAMPServer
+}
+
+func main() {
+	progressServer := NewProgressServer()
+	err := progressServer.Start()
+	if err != nil {
+		panic(err)
+	}
+
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, os.Interrupt)
+	<-interrupt
+
+	progressServer.Stop()
+}
+
+func NewProgressServer() *ProgressServer {
+	return &ProgressServer{
+		server: server.New(nil),
+	}
+}
+
+func (p *ProgressServer) Start() error {
+	return p.server.Start(server.StartSettings{
+		Settings: server.Settings{
+			Callbacks: server.CallbacksStruct{
+				OnConnectingFunc: p.onConnecting,
+			},
+
+			EnableCompression:  false,
+			CustomCapabilities: []string{openTelemetryCollectorReceiverAWSS3},
+		},
+		ListenEndpoint: "localhost:8080",
+		ListenPath:     "",
+		TLSConfig:      nil,
+		HTTPMiddleware: nil,
+	})
+}
+
+func (p *ProgressServer) Stop() {
+	p.server.Stop(context.Background())
+}
+
+func (p *ProgressServer) onConnecting(request *http.Request) types.ConnectionResponse {
+	fmt.Println("OnConnecting")
+	return types.ConnectionResponse{
+		Accept: true,
+		ConnectionCallbacks: server.ConnectionCallbacksStruct{
+			OnMessageFunc: p.onMessage,
+		},
+	}
+}
+
+func (p *ProgressServer) onMessage(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+	response := &protobufs.ServerToAgent{
+		InstanceUid: message.InstanceUid,
+	}
+
+	if message.CustomCapabilities != nil {
+		capabilities := make([]string, 0)
+		for _, capability := range message.CustomCapabilities.Capabilities {
+			if capability == "org.opentelemetry.collector.receiver.awss3" {
+
+				capabilities = append(capabilities, capability)
+			}
+		}
+		if len(capabilities) == 0 {
+			fmt.Println("🛑 - Agent does not support AWS S3 receiver progress")
+		} else {
+			fmt.Println("✅ - Agent supports AWS S3 receiver progress")
+		}
+
+		response.CustomCapabilities = &protobufs.CustomCapabilities{
+			Capabilities: capabilities,
+		}
+	}
+
+	if message.CustomMessage != nil && message.CustomMessage.Capability == openTelemetryCollectorReceiverAWSS3 {
+		if message.CustomMessage.Type == "TimeBasedIngestStatus" {
+			status := &TimeBasedIngestStatus{}
+			err := json.Unmarshal(message.CustomMessage.Data, status)
+			if err != nil {
+				fmt.Println("💣 - Error unmarshalling custom message data", err)
+			} else {
+				switch status.IngestStatus {
+				case "complete":
+					fmt.Println("🎉 - Ingest complete")
+				case "failed":
+					fmt.Println("🚨 - Ingest failed:", status.FailureMessage)
+				case "ingesting":
+					done := status.IngestTime.Sub(status.StartTime)
+					left := status.EndTime.Sub(status.IngestTime)
+					fmt.Printf("🚀 - Ingesting %s done, %s left (current %s)\n", done, left, status.IngestTime)
+				}
+			}
+		}
+	}
+
+	return response
+}

receiver/awss3receiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.27.11
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.16.15
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.53.1
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages v0.0.0-20240513080536-a133a8efefbe
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.100.0
 	go.opentelemetry.io/collector/confmap v0.100.0
@@ -54,6 +55,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/open-telemetry/opamp-go v0.14.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.19.0 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
@@ -72,3 +74,5 @@ require (
 	google.golang.org/protobuf v1.34.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages => ../../extension/opampcustommessages

receiver/awss3receiver/notifications.go

@@ -0,0 +1,91 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package awss3receiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awss3receiver"
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages"
+)
+
+const (
+	IngestStatusCompleted = "completed"
+	IngestStatusFailed    = "failed"
+	IngestStatusIngesting = "ingesting"
+	CustomCapability      = "org.opentelemetry.collector.receiver.awss3"
+)
+
+type StatusNotification struct {
+	TelemetryType  string    `json:"telemetry_type"`
+	IngestStatus   string    `json:"ingest_status"`
+	StartTime      time.Time `json:"start_time"`
+	EndTime        time.Time `json:"end_time"`
+	IngestTime     time.Time `json:"ingest_time"`
+	FailureMessage string    `json:"failure_message,omitempty"`
+}
+
+type statusNotifier interface {
+	Start(ctx context.Context, host component.Host) error
+	Shutdown(ctx context.Context) error
+	SendStatus(ctx context.Context, message StatusNotification)
+}
+
+type opampNotifier struct {
+	opampExtensionID component.ID
+	handler          opampcustommessages.CustomCapabilityHandler
+}
+
+func newNotifier(config *Config) statusNotifier {
+	if config.Notifications.OpAMP != nil {
+		return &opampNotifier{opampExtensionID: *config.Notifications.OpAMP}
+	}
+	return nil
+}
+
+func (n *opampNotifier) Start(_ context.Context, host component.Host) error {
+	ext, ok := host.GetExtensions()[n.opampExtensionID]
+	if !ok {
+		return fmt.Errorf("extension %q does not exist", n.opampExtensionID)
+	}
+
+	registry, ok := ext.(opampcustommessages.CustomCapabilityRegistry)
+	if !ok {
+		return fmt.Errorf("extension %q is not a custom message registry", n.opampExtensionID)
+	}
+
+	handler, err := registry.Register(CustomCapability)
+	if err != nil {
+		return fmt.Errorf("failed to register custom capability: %w", err)
+	}
+	n.handler = handler
+	return nil
+}
+
+func (n *opampNotifier) Shutdown(_ context.Context) error {
+	n.handler.Unregister()
+	return nil
+}
+
+func (n *opampNotifier) SendStatus(_ context.Context, message StatusNotification) {
+	bytes, err := json.Marshal(message)
+	if err != nil {
+		return
+	}
+	sendingChan, err := n.handler.SendMessage("TimeBasedIngestStatus", bytes)
+	switch {
+	case err == nil:
+		break
+	case errors.Is(err, types.ErrCustomMessagePending):
+		<-sendingChan
+	default:
+		return
+	}
+}