Update instance connectors on a connector reconcile (#4992)

* UpdateInstanceWithRillYAML on connector reconcile

* Simplified connector reconcile

* Merged main

* Fixed test

* fmt fix

* ResetState

proto/gen/rill/admin/v1/admin.swagger.yaml

@@ -2289,7 +2289,7 @@ definitions:
       `NullValue` is a singleton enumeration to represent the null value for the
       `Value` type union.
 
-       The JSON representation for `NullValue` is JSON `null`.
+      The JSON representation for `NullValue` is JSON `null`.
 
        - NULL_VALUE: Null value.
   rpcStatus:

proto/gen/rill/runtime/v1/runtime.swagger.yaml

@@ -2868,7 +2868,7 @@ definitions:
       `NullValue` is a singleton enumeration to represent the null value for the
       `Value` type union.
 
-       The JSON representation for `NullValue` is JSON `null`.
+      The JSON representation for `NullValue` is JSON `null`.
 
        - NULL_VALUE: Null value.
   rpcStatus:
@@ -3574,6 +3574,9 @@ definitions:
           NOTE : properties_from_variables and properties both should be used to get all properties.
   v1ConnectorState:
     type: object
+    properties:
+      specHash:
+        type: string
   v1ConnectorV2:
     type: object
     properties:

proto/rill/runtime/v1/resources.proto

@@ -537,6 +537,7 @@ message ConnectorSpec {
 }
 
 message ConnectorState {
+  string spec_hash = 1;
 }
 
 message ConnectorV2 {

runtime/compilers/rillv1/connectors.go

@@ -104,10 +104,15 @@ func (a *connectorAnalyzer) analyzeSource(ctx context.Context, r *Resource) {
 	spec := r.SourceSpec
 	srcProps := spec.Properties.AsMap()
 	_, sourceDriver, driverErr := a.parser.driverForConnector(spec.SourceConnector)
+	if driverErr != nil {
+		// Track the errored source connector and return
+		a.trackConnector(spec.SourceConnector, r, false)
+		return
+	}
 
 	// Check if we have anonymous access (unless we already know that we don't)
 	var anonAccess bool
-	if res, ok := a.result[spec.SourceConnector]; (!ok || res.AnonymousAccess) && driverErr != nil {
+	if res, ok := a.result[spec.SourceConnector]; !ok || res.AnonymousAccess {
 		anonAccess, _ = sourceDriver.HasAnonymousSourceAccess(ctx, srcProps, zap.NewNop())
 	}
 
@@ -133,10 +138,15 @@ func (a *connectorAnalyzer) analyzeModel(ctx context.Context, r *Resource) {
 	spec := r.ModelSpec
 	inputProps := spec.InputProperties.AsMap()
 	_, inputDriver, driverErr := a.parser.driverForConnector(spec.InputConnector)
+	if driverErr != nil {
+		// Track the errored input connector and return
+		a.trackConnector(spec.InputConnector, r, false)
+		return
+	}
 
 	// Check if we have anonymous access (unless we already know that we don't)
 	var anonAccess bool
-	if res, ok := a.result[spec.InputConnector]; (!ok || res.AnonymousAccess) && driverErr != nil {
+	if res, ok := a.result[spec.InputConnector]; !ok || res.AnonymousAccess {
 		anonAccess, _ = inputDriver.HasAnonymousSourceAccess(ctx, inputProps, zap.NewNop())
 	}
 
@@ -194,8 +204,6 @@ func (a *connectorAnalyzer) analyzeResourceNotifiers(r *Resource, notifiers []*r
 func (a *connectorAnalyzer) trackConnector(connector string, r *Resource, anonAccess bool) {
 	res, ok := a.result[connector]
 	if !ok {
-		driver, driverConnector, driverErr := a.parser.driverForConnector(connector)
-
 		// Search rill.yaml for default config properties for this connector
 		var defaultConfig map[string]string
 		if a.parser.RillYAML != nil {
@@ -215,19 +223,22 @@ func (a *connectorAnalyzer) trackConnector(connector string, r *Resource, anonAc
 			}
 		}
 
-		var driverSpec *drivers.Spec
-		if driverConnector != nil {
-			spec := driverConnector.Spec()
-			driverSpec = &spec
-		}
-
-		res = &Connector{
-			Name:            connector,
-			Driver:          driver,
-			Spec:            driverSpec,
-			DefaultConfig:   defaultConfig,
-			AnonymousAccess: true,
-			Err:             driverErr,
+		driver, driverConnector, driverErr := a.parser.driverForConnector(connector)
+		if driverErr != nil {
+			res = &Connector{
+				Name: connector,
+				Err:  driverErr,
+			}
+		} else {
+			driverSpec := driverConnector.Spec()
+			res = &Connector{
+				Name:            connector,
+				Driver:          driver,
+				Spec:            &driverSpec,
+				DefaultConfig:   defaultConfig,
+				AnonymousAccess: true,
+				Err:             driverErr,
+			}
 		}
 
 		a.result[connector] = res

runtime/controller_test.go

@@ -1390,7 +1390,8 @@ name: my-gcs
 		},
 		Resource: &runtimev1.Resource_Connector{
 			Connector: &runtimev1.ConnectorV2{
-				Spec: &runtimev1.ConnectorSpec{Driver: "s3", Properties: map[string]string{"region": "us-west-2"}},
+				Spec:  &runtimev1.ConnectorSpec{Driver: "s3", Properties: map[string]string{"region": "us-west-2"}},
+				State: &runtimev1.ConnectorState{},
 			},
 		},
 	})
@@ -1402,7 +1403,8 @@ name: my-gcs
 		},
 		Resource: &runtimev1.Resource_Connector{
 			Connector: &runtimev1.ConnectorV2{
-				Spec: &runtimev1.ConnectorSpec{Driver: "gcs"},
+				Spec:  &runtimev1.ConnectorSpec{Driver: "gcs"},
+				State: &runtimev1.ConnectorState{},
 			},
 		},
 	})

runtime/reconcilers/connector.go

@@ -2,10 +2,13 @@ package reconcilers
 
 import (
 	"context"
+	"crypto/md5"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"sort"
 
-	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
+	"github.com/rilldata/rill/proto/gen/rill/runtime/v1"
 	"github.com/rilldata/rill/runtime"
 )
 
@@ -46,6 +49,7 @@ func (r *ConnectorReconciler) AssignState(from, to *runtimev1.Resource) error {
 }
 
 func (r *ConnectorReconciler) ResetState(res *runtimev1.Resource) error {
+	res.GetConnector().State = &runtimev1.ConnectorState{}
 	return nil
 }
 
@@ -61,8 +65,65 @@ func (r *ConnectorReconciler) Reconcile(ctx context.Context, n *runtimev1.Resour
 
 	// Exit early for deletion
 	if self.Meta.DeletedOn != nil {
+		err = r.C.Runtime.UpdateInstanceConnector(ctx, r.C.InstanceID, self.Meta.Name.Name, nil)
+		if err != nil {
+			return runtime.ReconcileResult{Err: err}
+		}
 		return runtime.ReconcileResult{}
 	}
 
+	// Check if the spec has changed
+	specHash, err := r.executionSpecHash(t.Spec)
+	if err != nil {
+		return runtime.ReconcileResult{Err: err}
+	}
+
+	if specHash == t.State.SpecHash {
+		return runtime.ReconcileResult{}
+	}
+
+	// Update instance connectors
+	err = r.C.Runtime.UpdateInstanceConnector(ctx, r.C.InstanceID, self.Meta.Name.Name, t.Spec)
+	if err != nil {
+		return runtime.ReconcileResult{Err: err}
+	}
+
+	t.State.SpecHash = specHash
+
+	err = r.C.UpdateState(ctx, self.Meta.Name, self)
+	if err != nil {
+		return runtime.ReconcileResult{Err: err}
+	}
+
 	return runtime.ReconcileResult{}
 }
+
+func (r *ConnectorReconciler) executionSpecHash(spec *runtimev1.ConnectorSpec) (string, error) {
+	hash := md5.New()
+
+	_, err := hash.Write([]byte(spec.Driver))
+	if err != nil {
+		return "", err
+	}
+
+	// sort properties by key
+	keys := make([]string, 0, len(spec.Properties))
+	for k := range spec.Properties {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	// write properties to hash
+	for _, k := range keys {
+		_, err = hash.Write([]byte(k))
+		if err != nil {
+			return "", err
+		}
+		_, err = hash.Write([]byte(spec.Properties[k]))
+		if err != nil {
+			return "", err
+		}
+	}
+
+	return hex.EncodeToString(hash.Sum(nil)), nil
+}

runtime/runtime.go

@@ -167,6 +167,37 @@ func (r *Runtime) UpdateInstanceWithRillYAML(ctx context.Context, instanceID str
 	return r.EditInstance(ctx, inst, restartController)
 }
 
+// UpdateInstanceConnector upserts or removes a connector from an instance
+// If connector is nil, the connector is removed; otherwise, it is upserted
+func (r *Runtime) UpdateInstanceConnector(ctx context.Context, instanceID, name string, connector *runtimev1.ConnectorSpec) error {
+	inst, err := r.Instance(ctx, instanceID)
+	if err != nil {
+		return err
+	}
+
+	// remove the connector if it exists
+	for i, c := range inst.ProjectConnectors {
+		if c.Name == name {
+			inst.ProjectConnectors = append(inst.ProjectConnectors[:i], inst.ProjectConnectors[i+1:]...)
+			break
+		}
+	}
+
+	if connector == nil {
+		// connector should be removed
+		return r.EditInstance(ctx, inst, false)
+	}
+
+	// append the new/updated connector
+	inst.ProjectConnectors = append(inst.ProjectConnectors, &runtimev1.Connector{
+		Name:   name,
+		Type:   connector.Driver,
+		Config: connector.Properties,
+	})
+
+	return r.EditInstance(ctx, inst, false)
+}
+
 func instanceAnnotationsToAttribs(instance *drivers.Instance) []attribute.KeyValue {
 	attrs := make([]attribute.KeyValue, 0, len(instance.Annotations)+1)
 	attrs = append(attrs, attribute.String("instance_id", instance.ID))

runtime/testruntime/reconcile.go

@@ -171,6 +171,9 @@ func RequireResource(t testing.TB, rt *runtime.Runtime, id string, a *runtimev1.
 			e.StartedOn = nil
 			e.FinishedOn = nil
 		}
+	case runtime.ResourceKindConnector:
+		state := b.GetConnector().State
+		state.SpecHash = ""
 	}
 
 	// Hack to only compare the Resource field (not Meta)

web-admin/src/client/gen/index.schemas.ts

@@ -915,7 +915,7 @@ export interface RpcStatus {
  * `NullValue` is a singleton enumeration to represent the null value for the
 `Value` type union.
 
- The JSON representation for `NullValue` is JSON `null`.
+The JSON representation for `NullValue` is JSON `null`.
 
  - NULL_VALUE: Null value.
  */