feat: Add events recording

Author: jsbroks

Dockerfile

@@ -47,4 +47,6 @@ COPY --from=manager-builder /workspace/manager .
 COPY --from=console-builder /workspace/console/dist ./console
 USER 65532:65532
 
+ENV DEPLOYER_API_URL=https://deploy.wandb.ai/api
+
 ENTRYPOINT ["/manager"]

controllers/config.go

@@ -6,42 +6,55 @@ import (
 
 	apiv1 "github.com/wandb/operator/api/v1"
 	"github.com/wandb/operator/pkg/wandb/cdk8s/release"
+	"github.com/wandb/operator/pkg/wandb/status"
+	corev1 "k8s.io/api/core/v1"
 	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 func (r *WeightsAndBiasesReconciler) applyConfig(
 	ctx context.Context,
 	wandb *apiv1.WeightsAndBiases,
+	statusManager *status.Manager,
 	rel release.Release,
 	cfg map[string]interface{},
 ) error {
 	log := ctrllog.FromContext(ctx)
 
+	statusManager.Set(status.Downloading)
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "Downloading", "Downloading "+rel.Version())
 	log.Info("Downloading release", "version", rel.Version())
 	if err := rel.Download(); err != nil {
 		log.Error(err, "Failed to download release")
 		return err
 	}
 
+	statusManager.Set(status.Installing)
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "Installing", "Installing "+rel.Version())
 	log.Info("Install upgrade release", "version", rel.Version())
 	if err := rel.Install(); err != nil {
 		fmt.Println(err)
 		log.Error(err, "Failed to install release")
 		return err
 	}
 
+	statusManager.Set(status.Generating)
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "Generating", "Generating manifests "+rel.Version())
 	log.Info("Generating upgrade", "version", rel.Version())
 	if err := rel.Generate(cfg); err != nil {
 		fmt.Println(err)
 		log.Error(err, "Failed to generate release")
 		return err
 	}
 
+	statusManager.Set(status.Applying)
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "Applying", "Applying manifests "+rel.Version())
 	if err := rel.Apply(ctx, r.Client, wandb, r.Scheme); err != nil {
 		fmt.Println(err)
 		log.Error(err, "Failed to apply config")
 		return err
 	}
 
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "AppliedSuccessfully", "Successfully applied "+rel.Version())
+
 	return nil
 }

controllers/weightsandbiases_controller.go

@@ -18,11 +18,14 @@ package controllers
 
 import (
 	"context"
+	"reflect"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
@@ -39,7 +42,8 @@ import (
 // WeightsAndBiasesReconciler reconciles a WeightsAndBiases object
 type WeightsAndBiasesReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
+	Scheme   *runtime.Scheme
+	Recorder record.EventRecorder
 }
 
 //+kubebuilder:rbac:groups=apps.wandb.com,resources=weightsandbiases,verbs=get;list;watch;create;update;patch;delete
@@ -77,7 +81,8 @@ func (r *WeightsAndBiasesReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		log.Error(err, "Failed to set status")
 	}
 
-	statusManager.Set(status.Loading)
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "Reconciling", "Reconciling")
+
 	configManager := config.NewManager(ctx, r.Client, wandb, r.Scheme)
 	usersConfig, err := configManager.GetDesiredState()
 
@@ -93,22 +98,25 @@ func (r *WeightsAndBiasesReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		license = wandb.Spec.License
 	}
 
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "LoadingConfig", "Loading desired configuration")
 	// Apply configs in least to most priority order
 	desiredState := config.Merge(
 		usersConfig,
 		cdk8s.Github(),
-		cdk8s.Deployment(license),
+		cdk8s.Deployer(license),
 		cdk8s.Operator(wandb, r.Scheme),
 	)
 
 	if err != nil {
 		if errors.IsNotFound(err) {
+			r.Recorder.Event(wandb, corev1.EventTypeNormal, "CreatingConfig", "Creating config")
 			log.Info("No config found. Creating config...", "name", configManager.LatestConfigName())
 			usersConfig, err = configManager.CreateLatest(
 				desiredState.Release,
 				map[string]interface{}{},
 			)
 			if err != nil {
+				statusManager.Set(status.InvalidConfig)
 				log.Error(err, "Failed to create config")
 				return ctrlqueue.Requeue(err)
 			}
@@ -118,24 +126,29 @@ func (r *WeightsAndBiasesReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		}
 	}
 
+	r.Recorder.Event(wandb, corev1.EventTypeNormal, "ConfigBackup", "Creating a copy of config")
 	configManager.BackupLatest()
 
 	log.Info("Config values:", "version", desiredState.Release.Version(), "config", desiredState.Config)
 	if err != nil {
+		statusManager.Set(status.InvalidConfig)
 		log.Error(err, "Failed to get config")
 		return ctrlqueue.Requeue(err)
 	}
 
 	statusManager.Set(status.Loading)
 	log.Info("Applying config changes...", "version", usersConfig.Release.Version())
-	if err := r.applyConfig(ctx, wandb, usersConfig.Release, desiredState.Config); err != nil {
+	if err := r.applyConfig(ctx, wandb, statusManager, usersConfig.Release, desiredState.Config); err != nil {
+		statusManager.Set(status.InvalidConfig)
+		r.Recorder.Event(wandb, corev1.EventTypeNormal, "ApplyFailed", "Valid to apply config changes")
 		// TODO: Implement rollback
 		log.Error(err, "Failed to apply config changes.")
 		return ctrlqueue.DoNotRequeue()
 	}
 	log.Info("Successfully applied config", "version", usersConfig.Release.Version())
 
 	if desiredState.Release.Version() != usersConfig.Release.Version() {
+		r.Recorder.Event(wandb, corev1.EventTypeNormal, "VersionChangeDetected", "Version change detected ("+usersConfig.Release.Version()+" -> "+desiredState.Release.Version()+")")
 		log.Info(
 			"Version changed. Applying...",
 			"current", usersConfig.Release.Version(),
@@ -145,13 +158,15 @@ func (r *WeightsAndBiasesReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		statusManager.Set(status.Loading)
 		usersConfig.SetRelease(desiredState.Release)
 
-		if err := r.applyConfig(ctx, wandb, usersConfig.Release, desiredState.Config); err != nil {
+		if err := r.applyConfig(ctx, wandb, statusManager, usersConfig.Release, desiredState.Config); err != nil {
+			statusManager.Set(status.InvalidVersion)
+			r.Recorder.Event(wandb, corev1.EventTypeNormal, "VersionApplyFailed", "Valid to apply version changes")
 			// TODO: Implement rollback
 			log.Error(err, "Failed to upgrade to new version.")
 			return ctrlqueue.DoNotRequeue()
 		}
 
-		// Only if succesful do we save the state to a config map.
+		// Only if successful do we save the state to a config map.
 		if err := configManager.SetDesiredState(usersConfig); err != nil {
 			log.Error(err, "Failed to set desired state")
 			return ctrlqueue.DoNotRequeue()
@@ -168,9 +183,25 @@ func (r *WeightsAndBiasesReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	builder := ctrl.NewControllerManagedBy(mgr).
 		For(
 			&apiv1.WeightsAndBiases{},
-			builder.WithPredicates(predicate.GenerationChangedPredicate{}),
+			builder.WithPredicates(annotationChangedPredicate{}),
 		).
 		Owns(&corev1.Secret{}).
 		Owns(&corev1.ConfigMap{})
 	return builder.Complete(r)
 }
+
+// annotationChangedPredicate implements the Predicate interface. It is used to
+// watch changes in the annotation of the Custom Resource.
+type annotationChangedPredicate struct {
+	predicate.Funcs
+}
+
+// It checks if there is an update event and only returns true if the
+// annotations have changed. We use annotations to trigger a reconcile
+// functions.
+func (a annotationChangedPredicate) Update(e event.UpdateEvent) bool {
+	return !reflect.DeepEqual(
+		e.ObjectOld.GetAnnotations(),
+		e.ObjectNew.GetAnnotations(),
+	)
+}

main.go

@@ -91,8 +91,9 @@ func main() {
 	}
 
 	if err = (&controllers.WeightsAndBiasesReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Recorder: mgr.GetEventRecorderFor("weightsandbiases"),
+		Client:   mgr.GetClient(),
+		Scheme:   mgr.GetScheme(),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "WeightsAndBiases")
 		os.Exit(1)

pkg/utils/envs.go

@@ -0,0 +1,11 @@
+package utils
+
+import "os"
+
+func Getenv(key string, defaultValue string) string {
+	value := os.Getenv(key)
+	if value == "" {
+		return defaultValue
+	}
+	return value
+}

pkg/wandb/cdk8s/config/modifier.go

@@ -1,20 +1,29 @@
 package config
 
+// Modifier is used to override or recommend a config for the user. It will
+// modify or add defaults to the users config.
 type Modifier interface {
-	Recommend() *Config
-	Override() *Config
+	// Recommend will return a config that will be applied for the users config.
+	// This means the user can override any values set here cfg is the current
+	// state of the config before applying these changes
+	Recommend(cfg *Config) *Config
+	// Override will return a config that will be applied and override the users
+	// specified config. cfg is the current state of the config before applying
+	// these changes
+	Override(cfg *Config) *Config
 }
 
 // Merge will combine all the configs from the channels passed in. Last channel
-// will take precedent. If a channel returns nil for recommend or override, it will be skipped
+// will take precedent. If a channel returns nil for recommend or override, it
+// will be skipped
 func Merge(inputConfig *Config, providers ...Modifier) *Config {
 	cfg := &Config{Config: map[string]interface{}{}}
 
 	for _, channel := range providers {
 		if channel == nil {
 			continue
 		}
-		if c := channel.Recommend(); c != nil {
+		if c := channel.Recommend(cfg); c != nil {
 			cfg.Merge(c)
 		}
 	}
@@ -27,7 +36,7 @@ func Merge(inputConfig *Config, providers ...Modifier) *Config {
 		if channel == nil {
 			continue
 		}
-		if c := channel.Override(); c != nil {
+		if c := channel.Override(cfg); c != nil {
 			cfg.Merge(c)
 		}
 	}

pkg/wandb/cdk8s/deployer.go

@@ -4,7 +4,24 @@ import (
 	"github.com/wandb/operator/pkg/wandb/cdk8s/config"
 )
 
-// Deployment returns the config suggested by deployer
-func Deployment(license string) config.Modifier {
-	return nil
+// Deployer returns the config suggested by deployer
+func Deployer(license string) config.Modifier {
+	return &deployerChannel{}
+}
+
+type deployerResponse struct {
+	recommend *config.Config
+	override  *config.Config
+}
+
+type deployerChannel struct {
+	response deployerResponse
+}
+
+func (c deployerChannel) Recommend(_ *config.Config) *config.Config {
+	return c.response.recommend
+}
+
+func (c deployerChannel) Override(_ *config.Config) *config.Config {
+	return c.response.override
 }

pkg/wandb/cdk8s/operator.go

@@ -13,7 +13,7 @@ type operatorChannel struct {
 	scheme *runtime.Scheme
 }
 
-func (c operatorChannel) Recommend() *config.Config {
+func (c operatorChannel) Recommend(_ *config.Config) *config.Config {
 	gvk, _ := apiutil.GVKForObject(c.wandb, c.scheme)
 	return &config.Config{
 		Config: map[string]interface{}{
@@ -39,7 +39,7 @@ func (c operatorChannel) Recommend() *config.Config {
 	}
 }
 
-func (c operatorChannel) Override() *config.Config {
+func (c operatorChannel) Override(_ *config.Config) *config.Config {
 	cfg := &config.Config{}
 
 	cfg.SetConfig(c.wandb.Spec.Config.Object)