Skip to content

feat: use Server-Side Apply for CRD management #201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gfyrag wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat: use Server-Side Apply for CRD management #201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
45c8daa
feat: use Server-Side Apply for CRD management
gfyrag Jun 2, 2026
84af915
test: add error path tests for critical sync functions
gfyrag Jun 2, 2026
203eec7
test: replace mock-based tests with integration tests
gfyrag Jun 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 19 additions & 16 deletions internal/k8s_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand All @@ -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 {
Expand Down
113 changes: 66 additions & 47 deletions internal/membership_listener.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package internal

import (
"context"
"encoding/json"
"fmt"
"net/url"
"slices"
Expand All @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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
}
Expand All @@ -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
}

Expand All @@ -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

@flemzord flemzord Jul 15, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Existing CRs created by the pre-SSA agent have their fields owned by the inferred Go-http-client Update manager. If the first sync after upgrading changes one of those fields, this non-forced Apply conflicts, and returning the existing object as success permanently drops the desired update while the caller still reports the stack as updated. Please add a one-time ownership migration for resources known to have been created by the legacy agent, or otherwise surface and retry the conflict, plus an integration test that POSTs a legacy resource and then syncs a changed desired value.

}
return nil, errors.Wrap(err, "applying object")
}

return u, nil
return result, nil
}

func (c *membershipListener) createOrUpdateStackDependency(
Expand Down
122 changes: 122 additions & 0 deletions tests/membership_listener_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{}
Expand Down
Loading