diff --git a/internal/k8s_client.go b/internal/k8s_client.go index 3537201..1d1e763 100644 --- a/internal/k8s_client.go +++ b/internal/k8s_client.go @@ -2,6 +2,8 @@ package internal import ( "context" + "encoding/json" + "fmt" "github.com/formancehq/go-libs/v2/collectionutils" "github.com/formancehq/go-libs/v2/logging" @@ -19,8 +21,7 @@ import ( type K8SClient interface { Get(ctx context.Context, resource string, name string) (*unstructured.Unstructured, error) - Create(ctx context.Context, resource string, o *unstructured.Unstructured) error - Patch(ctx context.Context, resource, name string, body []byte) error + Apply(ctx context.Context, resource string, obj *unstructured.Unstructured, fieldManager string, force bool) (*unstructured.Unstructured, error) Delete(ctx context.Context, resource, name string) error EnsureNotExists(ctx context.Context, resource, name string) error EnsureNotExistsBySelector(ctx context.Context, resource string, selector labels.Selector) error @@ -43,22 +44,24 @@ func (c defaultK8SClient) Get(ctx context.Context, resource string, name string) return u, nil } -func (c defaultK8SClient) Create(ctx context.Context, resource string, o *unstructured.Unstructured) error { - return c.restClient. - Post(). - Resource(resource). - Body(o). - Do(ctx). - Into(o) -} - -func (c defaultK8SClient) Patch(ctx context.Context, resource, name string, body []byte) error { - return c.restClient.Patch(types.MergePatchType). - Name(name). - Body(body). +func (c defaultK8SClient) Apply(ctx context.Context, resource string, obj *unstructured.Unstructured, fieldManager string, force bool) (*unstructured.Unstructured, error) { + data, err := json.Marshal(obj.Object) + if err != nil { + return nil, err + } + result := &unstructured.Unstructured{} + err = c.restClient.Patch(types.ApplyPatchType). Resource(resource). + Name(obj.GetName()). + Param("fieldManager", fieldManager). + Param("force", fmt.Sprintf("%t", force)). + Body(data). Do(ctx). - Error() + Into(result) + if err != nil { + return nil, err + } + return result, nil } func (c defaultK8SClient) Delete(ctx context.Context, resource, name string) error { diff --git a/internal/membership_listener.go b/internal/membership_listener.go index b67a450..f82def1 100644 --- a/internal/membership_listener.go +++ b/internal/membership_listener.go @@ -3,7 +3,6 @@ package internal import ( "context" - "encoding/json" "fmt" "net/url" "slices" @@ -20,7 +19,6 @@ import ( "github.com/formancehq/stack/components/agent/internal/generated" "github.com/formancehq/stack/components/agent/internal/grpcclient" "github.com/pkg/errors" - "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -29,6 +27,11 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) +const ( + fieldManagerAgent = "formance-agent" + fieldManagerDisable = "formance-agent:disable" +) + //go:generate mockgen -source=membership_listener.go -destination=membership_client_generated.go -package=internal . MembershipClient type MembershipClient interface { Orders() chan *generated.Order @@ -363,7 +366,19 @@ func (c *membershipListener) deleteStack(ctx context.Context, stack *generated.D } func (c *membershipListener) disableStack(ctx context.Context, stack *generated.DisabledStack) { - if err := c.client.Patch(ctx, "Stacks", stack.ClusterName, []byte(`{"spec": {"disabled": true}}`)); err != nil { + obj := &unstructured.Unstructured{ + Object: map[string]any{ + "apiVersion": formanceGroupVersion.String(), + "kind": "Stack", + "metadata": map[string]any{ + "name": stack.ClusterName, + }, + "spec": map[string]any{ + "disabled": true, + }, + }, + } + if _, err := c.client.Apply(ctx, "Stacks", obj, fieldManagerDisable, true); err != nil { logging.FromContext(ctx).Errorf("Disabling cluster side: %s", err) return } @@ -372,8 +387,20 @@ func (c *membershipListener) disableStack(ctx context.Context, stack *generated. } func (c *membershipListener) enableStack(ctx context.Context, stack *generated.EnabledStack) { - if err := c.client.Patch(ctx, "Stacks", stack.ClusterName, []byte(`{"spec": {"disabled": false}}`)); err != nil { - logging.FromContext(ctx).Errorf("Disabling cluster side: %s", err) + obj := &unstructured.Unstructured{ + Object: map[string]any{ + "apiVersion": formanceGroupVersion.String(), + "kind": "Stack", + "metadata": map[string]any{ + "name": stack.ClusterName, + }, + "spec": map[string]any{ + "disabled": false, + }, + }, + } + if _, err := c.client.Apply(ctx, "Stacks", obj, fieldManagerDisable, true); err != nil { + logging.FromContext(ctx).Errorf("Enabling cluster side: %s", err) return } @@ -385,64 +412,56 @@ func (c *membershipListener) createOrUpdate(ctx context.Context, gvk schema.Grou logger := logging.FromContext(ctx).WithFields(map[string]any{ "gvk": gvk, }) - logger.Infof("creating object '%s'", name) + logger.Infof("applying object '%s'", name) + if content["metadata"] == nil { content["metadata"] = map[string]any{} } + md := content["metadata"].(map[string]any) - if content["metadata"].(map[string]any)["labels"] == nil { - content["metadata"].(map[string]any)["labels"] = map[string]any{} + if md["labels"] == nil { + md["labels"] = map[string]any{} + } + md["labels"].(map[string]any)["formance.com/created-by-agent"] = "true" + md["labels"].(map[string]any)["formance.com/stack"] = stackName + md["name"] = name + + if owner != nil { + md["ownerReferences"] = []any{ + map[string]any{ + "apiVersion": owner.APIVersion, + "kind": owner.Kind, + "name": owner.Name, + "uid": string(owner.UID), + }, + } } - content["metadata"].(map[string]any)["labels"].(map[string]any)["formance.com/created-by-agent"] = "true" - content["metadata"].(map[string]any)["labels"].(map[string]any)["formance.com/stack"] = stackName - content["metadata"].(map[string]any)["name"] = name + content["apiVersion"] = gvk.GroupVersion().String() + content["kind"] = gvk.Kind restMapping, err := c.restMapper.RESTMapping(gvk.GroupKind()) if err != nil { return nil, errors.Wrap(err, "getting rest mapping") } - u, err := c.client.Get(ctx, restMapping.Resource.Resource, name) - if err != nil { - if !apierrors.IsNotFound(err) { - return nil, errors.Wrap(err, "reading object") - } - - logger.Infof("Object not found, create a new one") - - u := &unstructured.Unstructured{} - u.SetUnstructuredContent(content) - u.SetGroupVersionKind(gvk) - u.SetName(name) - if owner != nil { - u.SetOwnerReferences([]metav1.OwnerReference{*owner}) - } - - if err := c.client.Create(ctx, restMapping.Resource.Resource, u); err != nil { - return nil, errors.Wrap(err, "creating object") - } - - return u, nil + u := &unstructured.Unstructured{} + u.SetUnstructuredContent(content) - } - - if equality.Semantic.DeepDerivative(content, u.Object) { - logger.Infof("Object found and has expected content, skip it") - return u, nil - } - - logger.Infof("Object exists and content differ, patch it") - contentData, err := json.Marshal(content) + result, err := c.client.Apply(ctx, restMapping.Resource.Resource, u, fieldManagerAgent, false) if err != nil { - return nil, err - } - - if err := c.client.Patch(ctx, restMapping.Resource.Resource, name, contentData); err != nil { - return nil, errors.Wrap(err, "patching object") + if apierrors.IsConflict(err) { + logger.Infof("Conflict applying %s/%s: field owned by another manager, skipping update", gvk.Kind, name) + existing, getErr := c.client.Get(ctx, restMapping.Resource.Resource, name) + if getErr != nil { + return nil, errors.Wrap(getErr, "getting existing object after conflict") + } + return existing, nil + } + return nil, errors.Wrap(err, "applying object") } - return u, nil + return result, nil } func (c *membershipListener) createOrUpdateStackDependency( diff --git a/tests/membership_listener_test.go b/tests/membership_listener_test.go index b2f48f9..fb3339b 100644 --- a/tests/membership_listener_test.go +++ b/tests/membership_listener_test.go @@ -227,6 +227,128 @@ var _ = Describe("Membership listener", func() { Expect(u).To(TargetStack(stack)) } }) + It("Should default versionsFromFile to 'default' when empty", func() { + s := &unstructured.Unstructured{} + Expect(LoadResource("Stacks", membershipStack.ClusterName, s)).To(Succeed()) + versionsFromFile, _, _ := unstructured.NestedString(s.Object, "spec", "versionsFromFile") + Expect(versionsFromFile).To(Equal("default")) + }) + It("Should use the SSA field manager 'formance-agent'", func() { + s := &unstructured.Unstructured{} + Expect(LoadResource("Stacks", membershipStack.ClusterName, s)).To(Succeed()) + + managedFields := s.GetManagedFields() + found := false + for _, mf := range managedFields { + if mf.Manager == "formance-agent" { + found = true + break + } + } + Expect(found).To(BeTrue(), "expected managedFields to contain 'formance-agent'") + }) + It("Should be idempotent on resync", func() { + s1 := &unstructured.Unstructured{} + Expect(LoadResource("Stacks", membershipStack.ClusterName, s1)).To(Succeed()) + rv1 := s1.GetResourceVersion() + + membershipClient.Orders() <- &generated.Order{ + Message: &generated.Order_ExistingStack{ + ExistingStack: membershipStack, + }, + } + + // Give the agent time to process the order + Consistently(func(g Gomega) { + s2 := &unstructured.Unstructured{} + g.Expect(LoadResource("Stacks", membershipStack.ClusterName, s2)).To(Succeed()) + g.Expect(s2.GetResourceVersion()).To(Equal(rv1)) + }, "2s", "200ms").Should(Succeed()) + }) + When("versions is set explicitly", func() { + BeforeEach(func() { + membershipStack.Versions = "v1.2.3" + membershipClient.Orders() <- &generated.Order{ + Message: &generated.Order_ExistingStack{ + ExistingStack: membershipStack, + }, + } + }) + It("Should use the explicit version", func() { + Eventually(func(g Gomega) { + s := &unstructured.Unstructured{} + g.Expect(LoadResource("Stacks", membershipStack.ClusterName, s)).To(Succeed()) + versionsFromFile, _, _ := unstructured.NestedString(s.Object, "spec", "versionsFromFile") + g.Expect(versionsFromFile).To(Equal("v1.2.3")) + }).Should(Succeed()) + }) + }) + When("a user manually edits a resource managed by the agent", func() { + It("Should preserve user-added annotations after resync", func() { + userPatch, err := json.Marshal(map[string]any{ + "metadata": map[string]any{ + "annotations": map[string]any{ + "user-custom/annotation": "user-value", + }, + }, + }) + Expect(err).To(BeNil()) + Expect(k8sClient.Patch(types.MergePatchType). + Resource("Stacks"). + Name(membershipStack.ClusterName). + Body(userPatch). + Do(context.Background()). + Error()).To(Succeed()) + + membershipClient.Orders() <- &generated.Order{ + Message: &generated.Order_ExistingStack{ + ExistingStack: membershipStack, + }, + } + + Eventually(func(g Gomega) { + s := &unstructured.Unstructured{} + g.Expect(LoadResource("Stacks", membershipStack.ClusterName, s)).To(Succeed()) + g.Expect(s.GetAnnotations()).To(HaveKeyWithValue("user-custom/annotation", "user-value")) + versionsFromFile, _, _ := unstructured.NestedString(s.Object, "spec", "versionsFromFile") + g.Expect(versionsFromFile).To(Equal("default")) + }).Should(Succeed()) + }) + + It("Should preserve user-added annotations on modules after resync", func() { + auth := &unstructured.Unstructured{} + Eventually(func() error { + return LoadResource("Auths", membershipStack.ClusterName, auth) + }).Should(BeNil()) + + userPatch, err := json.Marshal(map[string]any{ + "metadata": map[string]any{ + "annotations": map[string]any{ + "user-custom/note": "do not delete", + }, + }, + }) + Expect(err).To(BeNil()) + Expect(k8sClient.Patch(types.MergePatchType). + Resource("Auths"). + Name(membershipStack.ClusterName). + Body(userPatch). + Do(context.Background()). + Error()).To(Succeed()) + + membershipClient.Orders() <- &generated.Order{ + Message: &generated.Order_ExistingStack{ + ExistingStack: membershipStack, + }, + } + + Eventually(func(g Gomega) { + a := &unstructured.Unstructured{} + g.Expect(LoadResource("Auths", membershipStack.ClusterName, a)).To(Succeed()) + g.Expect(a.GetAnnotations()).To(HaveKeyWithValue("user-custom/note", "do not delete")) + }).Should(Succeed()) + }) + }) When("removing modules", func() { var ( modulesToRemove map[string]struct{} diff --git a/tests/versions_informer_test.go b/tests/versions_informer_test.go new file mode 100644 index 0000000..0677d0e --- /dev/null +++ b/tests/versions_informer_test.go @@ -0,0 +1,192 @@ +package tests + +import ( + "context" + "encoding/json" + "time" + + "github.com/formancehq/go-libs/v2/logging" + "github.com/formancehq/stack/components/agent/internal" + "github.com/google/uuid" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/dynamic" +) + +var _ = Describe("Versions informer", func() { + var ( + membershipClientMock *internal.MembershipClientMock + startListener func() + ) + BeforeEach(func() { + membershipClientMock = internal.NewMembershipClientMock() + dynamicClient, err := dynamic.NewForConfig(restConfig) + Expect(err).To(Succeed()) + + factory := internal.NewDynamicSharedInformerFactory(dynamicClient, 5*time.Minute) + Expect(internal.CreateVersionsInformer(factory, logging.Testing(), membershipClientMock)).To(Succeed()) + startListener = func() { + stopCh := make(chan struct{}) + factory.Start(stopCh) + DeferCleanup(func() { + close(stopCh) + }) + } + }) + When("a Versions resource is created", func() { + var version *unstructured.Unstructured + BeforeEach(func() { + version = &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": formanceGroupVersion.String(), + "kind": "Versions", + "metadata": map[string]interface{}{ + "name": uuid.NewString(), + "annotations": map[string]interface{}{ + "formance.com/deprecated": "true", + }, + }, + "spec": map[string]interface{}{ + "ledger": "v2.0.0", + "payments": "v1.5.0", + }, + }, + } + Expect(k8sClient.Post(). + Resource("Versions"). + Body(version). + Do(context.Background()). + Into(version)).To(Succeed()) + + startListener() + + DeferCleanup(func() { + k8sClient.Delete(). + Resource("Versions"). + Name(version.GetName()). + Do(context.Background()) + }) + }) + It("Should send AddedVersion with spec and deprecated flag", func() { + Eventually(func(g Gomega) { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetAddedVersion(); msg != nil && msg.Name == version.GetName() { + g.Expect(msg.Versions["ledger"]).To(Equal("v2.0.0")) + g.Expect(msg.Versions["payments"]).To(Equal("v1.5.0")) + g.Expect(msg.Deprecated).To(BeTrue()) + return + } + } + g.Expect(false).To(BeTrue(), "AddedVersion message not found") + }).Should(Succeed()) + }) + When("the spec is updated", func() { + BeforeEach(func() { + // Wait for AddedVersion first + Eventually(func() bool { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetAddedVersion(); msg != nil && msg.Name == version.GetName() { + return true + } + } + return false + }).Should(BeTrue()) + + patch, err := json.Marshal(map[string]any{ + "spec": map[string]any{ + "ledger": "v2.1.0", + "payments": "v1.5.0", + }, + }) + Expect(err).To(BeNil()) + Expect(k8sClient.Patch(types.MergePatchType). + Resource("Versions"). + Name(version.GetName()). + Body(patch). + Do(context.Background()). + Error()).To(Succeed()) + }) + It("Should send UpdatedVersion", func() { + Eventually(func(g Gomega) { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetUpdatedVersion(); msg != nil && msg.Name == version.GetName() { + g.Expect(msg.Versions["ledger"]).To(Equal("v2.1.0")) + return + } + } + g.Expect(false).To(BeTrue(), "UpdatedVersion message not found") + }).Should(Succeed()) + }) + }) + When("the resource is deleted", func() { + BeforeEach(func() { + // Wait for AddedVersion first + Eventually(func() bool { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetAddedVersion(); msg != nil && msg.Name == version.GetName() { + return true + } + } + return false + }).Should(BeTrue()) + + Expect(k8sClient.Delete(). + Resource("Versions"). + Name(version.GetName()). + Do(context.Background()).Error()).To(Succeed()) + }) + It("Should send DeletedVersion", func() { + Eventually(func(g Gomega) { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetDeletedVersion(); msg != nil && msg.Name == version.GetName() { + return + } + } + g.Expect(false).To(BeTrue(), "DeletedVersion message not found") + }).Should(Succeed()) + }) + }) + }) + When("a Versions resource has no spec", func() { + var version *unstructured.Unstructured + BeforeEach(func() { + version = &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": formanceGroupVersion.String(), + "kind": "Versions", + "metadata": map[string]interface{}{ + "name": uuid.NewString(), + }, + }, + } + Expect(k8sClient.Post(). + Resource("Versions"). + Body(version). + Do(context.Background()). + Into(version)).To(Succeed()) + + startListener() + + DeferCleanup(func() { + k8sClient.Delete(). + Resource("Versions"). + Name(version.GetName()). + Do(context.Background()) + }) + }) + It("Should send AddedVersion with nil versions", func() { + Eventually(func(g Gomega) { + for _, message := range membershipClientMock.GetMessages() { + if msg := message.GetAddedVersion(); msg != nil && msg.Name == version.GetName() { + g.Expect(msg.Versions).To(BeEmpty()) + g.Expect(msg.Deprecated).To(BeFalse()) + return + } + } + g.Expect(false).To(BeTrue(), "AddedVersion message not found") + }).Should(Succeed()) + }) + }) +})