@@ -23,10 +23,12 @@ import (
2323"github.com/sorintlab/stolon/internal/cluster"
2424"github.com/sorintlab/stolon/internal/util"
2525
26+ jsonpatch"github.com/evanphx/json-patch"
2627v1"k8s.io/api/core/v1"
2728apierrors"k8s.io/apimachinery/pkg/api/errors"
2829metav1"k8s.io/apimachinery/pkg/apis/meta/v1"
2930"k8s.io/apimachinery/pkg/labels"
31+ "k8s.io/apimachinery/pkg/types"
3032"k8s.io/client-go/kubernetes"
3133"k8s.io/client-go/kubernetes/scheme"
3234v1core"k8s.io/client-go/kubernetes/typed/core/v1"
@@ -72,6 +74,43 @@ func (s *KubeStore) labelSelector(componentLabel ComponentLabelValue) labels.Sel
7274return labels .SelectorFromSet (selector )
7375}
7476
77+ func (s * KubeStore )patchKubeStatusAnnotation (annotationData []byte )error {
78+ podsClient := s .client .CoreV1 ().Pods (s .namespace )
79+ retryErr := retry .RetryOnConflict (retry .DefaultRetry ,func ()error {
80+ pod ,err := podsClient .Get (s .podName , metav1.GetOptions {})
81+ if err != nil {
82+ return fmt .Errorf ("failed to get latest version of pod: %v" ,err )
83+ }
84+
85+ oldPodJSON ,err := json .Marshal (pod )
86+ if err != nil {
87+ return fmt .Errorf ("failed to marshal pod: %v" ,err )
88+ }
89+
90+ if pod .Annotations == nil {
91+ pod .Annotations = map [string ]string {}
92+ }
93+ pod .Annotations [util .KubeStatusAnnnotation ]= string (annotationData )
94+
95+ newPodJSON ,err := json .Marshal (pod )
96+ if err != nil {
97+ return fmt .Errorf ("failed to marshal pod: %v" ,err )
98+ }
99+
100+ patchBytes ,err := jsonpatch .CreateMergePatch (oldPodJSON ,newPodJSON )
101+ if err != nil {
102+ return fmt .Errorf ("failed to create pod merge patch: %v" ,err )
103+ }
104+
105+ _ ,err = podsClient .Patch (s .podName ,types .MergePatchType ,patchBytes )
106+ return err
107+ })
108+ if retryErr != nil {
109+ return fmt .Errorf ("update failed: %v" ,retryErr )
110+ }
111+ return nil
112+ }
113+
75114func (s * KubeStore )AtomicPutClusterData (ctx context.Context ,cd * cluster.ClusterData ,previous * KVPair ) (* KVPair ,error ) {
76115cdj ,err := json .Marshal (cd )
77116if err != nil {
@@ -210,23 +249,7 @@ func (s *KubeStore) SetKeeperInfo(ctx context.Context, id string, ms *cluster.Ke
210249if err != nil {
211250return err
212251}
213- podsClient := s .client .CoreV1 ().Pods (s .namespace )
214- retryErr := retry .RetryOnConflict (retry .DefaultRetry ,func ()error {
215- result ,err := podsClient .Get (s .podName , metav1.GetOptions {})
216- if err != nil {
217- return fmt .Errorf ("failed to get latest version of pod: %v" ,err )
218- }
219- if result .Annotations == nil {
220- result .Annotations = map [string ]string {}
221- }
222- result .Annotations [util .KubeStatusAnnnotation ]= string (msj )
223- _ ,err = podsClient .Update (result )
224- return err
225- })
226- if retryErr != nil {
227- return fmt .Errorf ("update failed: %v" ,retryErr )
228- }
229- return nil
252+ return s .patchKubeStatusAnnotation (msj )
230253}
231254
232255func (s * KubeStore )GetKeepersInfo (ctx context.Context ) (cluster.KeepersInfo ,error ) {
@@ -261,23 +284,7 @@ func (s *KubeStore) SetSentinelInfo(ctx context.Context, si *cluster.SentinelInf
261284if err != nil {
262285return err
263286}
264- podsClient := s .client .CoreV1 ().Pods (s .namespace )
265- retryErr := retry .RetryOnConflict (retry .DefaultRetry ,func ()error {
266- result ,err := podsClient .Get (s .podName , metav1.GetOptions {})
267- if err != nil {
268- return fmt .Errorf ("failed to get latest version of pod: %v" ,err )
269- }
270- if result .Annotations == nil {
271- result .Annotations = map [string ]string {}
272- }
273- result .Annotations [util .KubeStatusAnnnotation ]= string (sij )
274- _ ,err = podsClient .Update (result )
275- return err
276- })
277- if retryErr != nil {
278- return fmt .Errorf ("update failed: %v" ,retryErr )
279- }
280- return nil
287+ return s .patchKubeStatusAnnotation (sij )
281288}
282289
283290func (s * KubeStore )GetSentinelsInfo (ctx context.Context ) (cluster.SentinelsInfo ,error ) {
@@ -312,23 +319,7 @@ func (s *KubeStore) SetProxyInfo(ctx context.Context, pi *cluster.ProxyInfo, ttl
312319if err != nil {
313320return err
314321}
315- podsClient := s .client .CoreV1 ().Pods (s .namespace )
316- retryErr := retry .RetryOnConflict (retry .DefaultRetry ,func ()error {
317- result ,err := podsClient .Get (s .podName , metav1.GetOptions {})
318- if err != nil {
319- return fmt .Errorf ("failed to get latest version of pod: %v" ,err )
320- }
321- if result .Annotations == nil {
322- result .Annotations = map [string ]string {}
323- }
324- result .Annotations [util .KubeStatusAnnnotation ]= string (pij )
325- _ ,err = podsClient .Update (result )
326- return err
327- })
328- if retryErr != nil {
329- return fmt .Errorf ("update failed: %v" ,retryErr )
330- }
331- return nil
322+ return s .patchKubeStatusAnnotation (pij )
332323}
333324
334325func (s * KubeStore )GetProxiesInfo (ctx context.Context ) (cluster.ProxiesInfo ,error ) {