Merge pull request #754 from jbw976/change-logs

Change logs support to track all changes being made to managed resources

apis/changelogs/proto/v1alpha1/changelog.proto

@@ -0,0 +1,88 @@
+/*
+Copyright 2024 The Crossplane 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.
+*/
+
+syntax = "proto3";
+
+import "google/protobuf/struct.proto";
+import "google/protobuf/timestamp.proto";
+
+// buf:lint:ignore PACKAGE_DIRECTORY_MATCH
+package changelogs.proto.v1alpha1;
+
+option go_package = "github.com/crossplane/crossplane-runtime/apis/changelogs/proto/v1alpha1";
+
+// ChangeLogService is a service that provides the ability to send change log
+// entries.
+service ChangeLogService {
+  // SendChangeLog sends a change log entry to the change log service.
+  rpc SendChangeLog (SendChangeLogRequest) returns (SendChangeLogResponse) {}
+}
+
+// SendChangeLogRequest represents a request to send a single change log entry.
+message SendChangeLogRequest {
+  // The change log entry to send as part of this request.
+  ChangeLogEntry entry = 1;
+}
+
+// ChangeLogEntry represents a single change log entry, with detailed information
+// about the resource that was changed.
+message ChangeLogEntry {
+  // The timestamp at which the change occurred.
+  google.protobuf.Timestamp timestamp = 1;
+
+  // The name and version of the provider that is making the change to the
+  // resource.
+  string provider = 2;
+
+  // The API version of the resource that was changed, e.g. Group/Version.
+  string api_version = 3;
+
+  // The kind of the resource that was changed.
+  string kind = 4;
+
+  // The name of the resource that was changed.
+  string name = 5;
+
+  // The external name of the resource that was changed.
+  string external_name = 6;
+
+  // The type of operation that was performed on the resource, e.g. Create,
+  // Update, or Delete.
+  OperationType operation = 7;
+
+  // A full snapshot of the resource's state, as observed directly before the
+  // resource was changed.
+  google.protobuf.Struct snapshot = 8;
+
+  // An optional error message that describes any error encountered while
+  // performing the operation on the resource.
+  optional string error_message = 9;
+
+  // An optional additional details that can be provided for further context
+  // about the change.
+  map<string, string> additional_details = 10;
+}
+
+// OperationType represents the type of operation that was performed on a
+// resource.
+enum OperationType {
+  OPERATION_TYPE_UNSPECIFIED = 0;
+  OPERATION_TYPE_CREATE = 1;
+  OPERATION_TYPE_UPDATE = 2;
+  OPERATION_TYPE_DELETE = 3;
+}
+
+// SendChangeLogResponse is the response returned by the ChangeLogService after
+// a change log entry is sent. Currently, this is an empty message as the only
+// useful information expected to sent back at this time will be through errors.
+message SendChangeLogResponse {}

go.mod

@@ -21,6 +21,7 @@ require (
 	k8s.io/client-go v0.30.0
 	k8s.io/component-base v0.30.0
 	k8s.io/klog/v2 v2.120.1
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b
 	sigs.k8s.io/controller-runtime v0.18.2
 	sigs.k8s.io/controller-tools v0.14.0
 	sigs.k8s.io/yaml v1.4.0
@@ -80,7 +81,6 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect
-	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )

pkg/controller/options.go

@@ -67,6 +67,9 @@ type Options struct {
 
 	// MetricOptions for recording metrics.
 	MetricOptions *MetricOptions
+
+	// ChangeLogOptions for recording change logs.
+	ChangeLogOptions *ChangeLogOptions
 }
 
 // ForControllerRuntime extracts options for controller-runtime.
@@ -97,3 +100,9 @@ type MetricOptions struct {
 	// MRStateMetrics to use for recording state metrics.
 	MRStateMetrics *statemetrics.MRStateMetrics
 }
+
+// ChangeLogOptions for recording changes to managed resources into the change
+// logs.
+type ChangeLogOptions struct {
+	ChangeLogger managed.ChangeLogger
+}

pkg/feature/features.go

@@ -20,3 +20,8 @@ package feature
 // Management Policies. See the below design for more details.
 // https://github.com/crossplane/crossplane/pull/3531
 const EnableBetaManagementPolicies Flag = "EnableBetaManagementPolicies"
+
+// EnableAlphaChangeLogs enables alpha support for capturing change logs during
+// reconciliation. See the following design for more details:
+// https://github.com/crossplane/crossplane/pull/5822
+const EnableAlphaChangeLogs Flag = "EnableAlphaChangeLogs"

pkg/reconciler/managed/changelogger.go

@@ -0,0 +1,135 @@
+/*
+Copyright 2024 The Crossplane 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 managed
+
+import (
+	"context"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"k8s.io/utils/ptr"
+
+	"github.com/crossplane/crossplane-runtime/apis/changelogs/proto/v1alpha1"
+	"github.com/crossplane/crossplane-runtime/pkg/errors"
+	"github.com/crossplane/crossplane-runtime/pkg/meta"
+	"github.com/crossplane/crossplane-runtime/pkg/resource"
+)
+
+const (
+	defaultSendTimeout = 10 * time.Second
+)
+
+// ChangeLogger is an interface for recording changes made to resources to the
+// change logs.
+type ChangeLogger interface {
+	Log(ctx context.Context, managed resource.Managed, opType v1alpha1.OperationType, changeErr error, ad AdditionalDetails) error
+}
+
+// GRPCChangeLogger processes changes to resources and helps to send them to the
+// change log gRPC service.
+type GRPCChangeLogger struct {
+	client          v1alpha1.ChangeLogServiceClient
+	providerVersion string
+	sendTimeout     time.Duration
+}
+
+// NewGRPCChangeLogger creates a new gRPC based ChangeLogger initialized with
+// the given client.
+func NewGRPCChangeLogger(client v1alpha1.ChangeLogServiceClient, o ...GRPCChangeLoggerOption) *GRPCChangeLogger {
+	g := &GRPCChangeLogger{
+		client:      client,
+		sendTimeout: defaultSendTimeout,
+	}
+
+	for _, clo := range o {
+		clo(g)
+	}
+
+	return g
+}
+
+// A GRPCChangeLoggerOption configures a GRPCChangeLoggerOption.
+type GRPCChangeLoggerOption func(*GRPCChangeLogger)
+
+// WithProviderVersion sets the provider version to be included in the change
+// log entry.
+func WithProviderVersion(version string) GRPCChangeLoggerOption {
+	return func(g *GRPCChangeLogger) {
+		g.providerVersion = version
+	}
+}
+
+// WithSendTimeout sets the timeout for sending and/or waiting for change log
+// entries to the change log service.
+func WithSendTimeout(timeout time.Duration) GRPCChangeLoggerOption {
+	return func(g *GRPCChangeLogger) {
+		g.sendTimeout = timeout
+	}
+}
+
+// Log sends the given change log entry to the change log service.
+func (g *GRPCChangeLogger) Log(ctx context.Context, managed resource.Managed, opType v1alpha1.OperationType, changeErr error, ad AdditionalDetails) error {
+	// get an error message from the error if it exists
+	var changeErrMessage *string
+	if changeErr != nil {
+		changeErrMessage = ptr.To(changeErr.Error())
+	}
+
+	// capture the full state of the managed resource from before we performed the change
+	snapshot, err := resource.AsProtobufStruct(managed)
+	if err != nil {
+		return errors.Wrap(err, "cannot snapshot managed resource")
+	}
+
+	gvk := managed.GetObjectKind().GroupVersionKind()
+
+	entry := &v1alpha1.ChangeLogEntry{
+		Timestamp:         timestamppb.Now(),
+		Provider:          g.providerVersion,
+		ApiVersion:        gvk.GroupVersion().String(),
+		Kind:              gvk.Kind,
+		Name:              managed.GetName(),
+		ExternalName:      meta.GetExternalName(managed),
+		Operation:         opType,
+		Snapshot:          snapshot,
+		ErrorMessage:      changeErrMessage,
+		AdditionalDetails: ad,
+	}
+
+	// create a specific context and timeout for sending the change log entry
+	// that is different than the parent context that is for the entire
+	// reconciliation
+	sendCtx, sendCancel := context.WithTimeout(ctx, g.sendTimeout)
+	defer sendCancel()
+
+	// send everything we've got to the change log service
+	_, err = g.client.SendChangeLog(sendCtx, &v1alpha1.SendChangeLogRequest{Entry: entry}, grpc.WaitForReady(true))
+	return errors.Wrap(err, "cannot send change log entry")
+}
+
+// nopChangeLogger does nothing for recording change logs, this is the default
+// implementation if a provider has not enabled the change logs feature.
+type nopChangeLogger struct{}
+
+func newNopChangeLogger() *nopChangeLogger {
+	return &nopChangeLogger{}
+}
+
+func (n *nopChangeLogger) Log(_ context.Context, _ resource.Managed, _ v1alpha1.OperationType, _ error, _ AdditionalDetails) error {
+	return nil
+}