mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-22 03:11:40 +00:00
Merge pull request #37038 from ymqytw/retry_old_patch_after_new_patch_fail
Automatic merge from submit-queue Fix kubectl Stratigic Merge Patch compatibility As @smarterclayton pointed out in [comment1](https://github.com/kubernetes/kubernetes/pull/35647#pullrequestreview-8290820) and [comment2](https://github.com/kubernetes/kubernetes/pull/35647#pullrequestreview-8290847) in PR #35647, we cannot assume the API servers publish version and they shares the same version. This PR removes all the calls of GetServerSupportedSMPatchVersion(). Change the behavior of `apply` and `edit` to: Retrying with the old patch version, if the new version fails. Default other usage of SMPatch to the new version, since they don't update list of primitives. fixes #36916 cc: @pwittrock @smarterclayton
This commit is contained in:
commit
b9d2d74a94
@ -244,9 +244,8 @@ func (e *eventLogger) eventObserve(newEvent *api.Event) (*api.Event, []byte, err
|
||||
|
||||
newData, _ := json.Marshal(event)
|
||||
oldData, _ := json.Marshal(eventCopy2)
|
||||
// TODO: need to figure out if we need to let eventObserve() use the new behavior of StrategicMergePatch.
|
||||
// Currently default to old behavior now. Ref: issue #35936
|
||||
patch, err = strategicpatch.CreateStrategicMergePatch(oldData, newData, event, strategicpatch.SMPatchVersion_1_0)
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since we only update Count and LastTimestamp, and none of them has list of primitives
|
||||
patch, err = strategicpatch.CreateStrategicMergePatch(oldData, newData, event, strategicpatch.SMPatchVersion_1_5)
|
||||
}
|
||||
|
||||
// record our new observation
|
||||
|
@ -59,10 +59,6 @@ type nodeStatusUpdater struct {
|
||||
}
|
||||
|
||||
func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
|
||||
smPatchVersion, err := strategicpatch.GetServerSupportedSMPatchVersion(nsu.kubeClient.Discovery())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
nodesToUpdate := nsu.actualStateOfWorld.GetVolumesToReportAttached()
|
||||
for nodeName, attachedVolumes := range nodesToUpdate {
|
||||
nodeObj, exists, err := nsu.nodeInformer.GetStore().GetByKey(string(nodeName))
|
||||
@ -111,8 +107,9 @@ func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
|
||||
err)
|
||||
}
|
||||
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since updateNodeStatus doesn't update any lists of primitives
|
||||
patchBytes, err :=
|
||||
strategicpatch.CreateStrategicMergePatch(oldData, newData, node, smPatchVersion)
|
||||
strategicpatch.CreateStrategicMergePatch(oldData, newData, node, strategicpatch.SMPatchVersion_1_5)
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"failed to CreateStrategicMergePatch for node %q. %v",
|
||||
|
@ -191,6 +191,7 @@ go_test(
|
||||
"//pkg/runtime/serializer/streaming:go_default_library",
|
||||
"//pkg/types:go_default_library",
|
||||
"//pkg/util/intstr:go_default_library",
|
||||
"//pkg/util/strategicpatch:go_default_library",
|
||||
"//pkg/util/strings:go_default_library",
|
||||
"//pkg/util/term:go_default_library",
|
||||
"//pkg/util/wait:go_default_library",
|
||||
|
@ -223,12 +223,6 @@ func (o AnnotateOptions) RunAnnotate(f cmdutil.Factory, cmd *cobra.Command) erro
|
||||
}
|
||||
outputObj = obj
|
||||
} else {
|
||||
// retrieves server version to determine which SMPatchVersion to use.
|
||||
smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
name, namespace := info.Name, info.Namespace
|
||||
oldData, err := json.Marshal(obj)
|
||||
if err != nil {
|
||||
@ -245,7 +239,8 @@ func (o AnnotateOptions) RunAnnotate(f cmdutil.Factory, cmd *cobra.Command) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, smPatchVersion)
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since it just update the annotation which is a map[string]string
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, strategicpatch.SMPatchVersion_1_5)
|
||||
createdPatch := err == nil
|
||||
if err != nil {
|
||||
glog.V(2).Infof("couldn't compute patch: %v", err)
|
||||
|
@ -195,11 +195,6 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *App
|
||||
visitedUids := sets.NewString()
|
||||
visitedNamespaces := sets.NewString()
|
||||
|
||||
smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
count := 0
|
||||
err = r.Visit(func(info *resource.Info, err error) error {
|
||||
// In this method, info.Object contains the object retrieved from the server
|
||||
@ -270,13 +265,13 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *App
|
||||
gracePeriod: options.GracePeriod,
|
||||
}
|
||||
|
||||
patchBytes, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name, smPatchVersion)
|
||||
patchBytes, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name)
|
||||
if err != nil {
|
||||
return cmdutil.AddSourceToErr(fmt.Sprintf("applying patch:\n%s\nto:\n%v\nfor:", patchBytes, info), info.Source, err)
|
||||
}
|
||||
|
||||
if cmdutil.ShouldRecord(cmd, info) {
|
||||
patch, err := cmdutil.ChangeResourcePatch(info, f.Command(), smPatchVersion)
|
||||
patch, err := cmdutil.ChangeResourcePatch(info, f.Command())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -512,7 +507,7 @@ type patcher struct {
|
||||
gracePeriod int
|
||||
}
|
||||
|
||||
func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string, smPatchVersion strategicpatch.StrategicMergePatchVersion) ([]byte, error) {
|
||||
func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string) ([]byte, error) {
|
||||
// Serialize the current configuration of the object from the server.
|
||||
current, err := runtime.Encode(p.encoder, obj)
|
||||
if err != nil {
|
||||
@ -536,20 +531,29 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
|
||||
}
|
||||
|
||||
// Compute a three way strategic merge patch to send to server.
|
||||
patch, err := strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite, smPatchVersion)
|
||||
|
||||
patch, err := strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite, strategicpatch.SMPatchVersion_1_5)
|
||||
// If creating a patch fails, retrying with SMPatchVersion_1_0 is not helpful. So we return the error.
|
||||
if err != nil {
|
||||
format := "creating patch with:\noriginal:\n%s\nmodified:\n%s\ncurrent:\n%s\nfor:"
|
||||
return nil, cmdutil.AddSourceToErr(fmt.Sprintf(format, original, modified, current), source, err)
|
||||
}
|
||||
|
||||
_, err = p.helper.Patch(namespace, name, api.StrategicMergePatchType, patch)
|
||||
if errors.IsInternalError(err) {
|
||||
// Retry SMPatchVersion_1_0 when applying the SMPatchVersion_1_5 patch returns an Internal Error (500).
|
||||
// Because the failure may be due to the server not supporting the SMPatchVersion_1_5 patch.
|
||||
patch, err = strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite, strategicpatch.SMPatchVersion_1_0)
|
||||
if err != nil {
|
||||
format := "creating patch with:\noriginal:\n%s\nmodified:\n%s\ncurrent:\n%s\nfor:"
|
||||
return nil, cmdutil.AddSourceToErr(fmt.Sprintf(format, original, modified, current), source, err)
|
||||
}
|
||||
_, err = p.helper.Patch(namespace, name, api.StrategicMergePatchType, patch)
|
||||
}
|
||||
return patch, err
|
||||
}
|
||||
|
||||
func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string, smPatchVersion strategicpatch.StrategicMergePatchVersion) ([]byte, error) {
|
||||
func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string) ([]byte, error) {
|
||||
var getErr error
|
||||
patchBytes, err := p.patchSimple(current, modified, source, namespace, name, smPatchVersion)
|
||||
patchBytes, err := p.patchSimple(current, modified, source, namespace, name)
|
||||
for i := 1; i <= maxPatchRetry && errors.IsConflict(err); i++ {
|
||||
if i > triesBeforeBackOff {
|
||||
p.backOff.Sleep(backOffPeriod)
|
||||
@ -558,7 +562,7 @@ func (p *patcher) patch(current runtime.Object, modified []byte, source, namespa
|
||||
if getErr != nil {
|
||||
return nil, getErr
|
||||
}
|
||||
patchBytes, err = p.patchSimple(current, modified, source, namespace, name, smPatchVersion)
|
||||
patchBytes, err = p.patchSimple(current, modified, source, namespace, name)
|
||||
}
|
||||
if err != nil && p.force {
|
||||
patchBytes, err = p.deleteAndCreate(modified, namespace, name)
|
||||
|
@ -23,6 +23,7 @@ import (
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
@ -37,6 +38,7 @@ import (
|
||||
cmdtesting "k8s.io/kubernetes/pkg/kubectl/cmd/testing"
|
||||
cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
"k8s.io/kubernetes/pkg/util/strategicpatch"
|
||||
)
|
||||
|
||||
func TestApplyExtraArgsFail(t *testing.T) {
|
||||
@ -142,6 +144,58 @@ func readAndAnnotateService(t *testing.T, filename string) (string, []byte) {
|
||||
return annotateRuntimeObject(t, svc1, svc2, "Service")
|
||||
}
|
||||
|
||||
func setFinalizersRuntimeObject(t *testing.T, originalObj, currentObj runtime.Object) (string, []byte) {
|
||||
originalAccessor, err := meta.Accessor(originalObj)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
originalFinalizers := []string{"a/a"}
|
||||
originalAccessor.SetFinalizers(originalFinalizers)
|
||||
original, err := runtime.Encode(testapi.Default.Codec(), originalObj)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
currentAccessor, err := meta.Accessor(currentObj)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
currentFinalizers := []string{"b/b"}
|
||||
currentAccessor.SetFinalizers(currentFinalizers)
|
||||
|
||||
currentAnnotations := currentAccessor.GetAnnotations()
|
||||
if currentAnnotations == nil {
|
||||
currentAnnotations = make(map[string]string)
|
||||
}
|
||||
currentAnnotations[annotations.LastAppliedConfigAnnotation] = string(original)
|
||||
currentAccessor.SetAnnotations(currentAnnotations)
|
||||
current, err := runtime.Encode(testapi.Default.Codec(), currentObj)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
return currentAccessor.GetName(), current
|
||||
}
|
||||
|
||||
func readAndSetFinalizersReplicationController(t *testing.T, filename string) (string, []byte) {
|
||||
rc1 := readReplicationControllerFromFile(t, filename)
|
||||
rc2 := readReplicationControllerFromFile(t, filename)
|
||||
name, rcBytes := setFinalizersRuntimeObject(t, rc1, rc2)
|
||||
return name, rcBytes
|
||||
}
|
||||
|
||||
func isSMPatchVersion_1_5(t *testing.T, req *http.Request) bool {
|
||||
patch, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// SMPatchVersion_1_5 patch should has string "mergeprimitiveslist"
|
||||
return strings.Contains(string(patch), strategicpatch.MergePrimitivesListDirective)
|
||||
}
|
||||
|
||||
func validatePatchApplication(t *testing.T, req *http.Request) {
|
||||
patch, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
@ -223,6 +277,65 @@ func TestApplyObject(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyRetryWithSMPatchVersion_1_5(t *testing.T) {
|
||||
initTestErrorHandler(t)
|
||||
nameRC, currentRC := readAndSetFinalizersReplicationController(t, filenameRC)
|
||||
pathRC := "/namespaces/test/replicationcontrollers/" + nameRC
|
||||
|
||||
firstPatch := true
|
||||
retry := false
|
||||
f, tf, _, ns := cmdtesting.NewAPIFactory()
|
||||
tf.Printer = &testPrinter{}
|
||||
tf.Client = &fake.RESTClient{
|
||||
NegotiatedSerializer: ns,
|
||||
Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
|
||||
switch p, m := req.URL.Path, req.Method; {
|
||||
case p == pathRC && m == "GET":
|
||||
bodyRC := ioutil.NopCloser(bytes.NewReader(currentRC))
|
||||
return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: bodyRC}, nil
|
||||
case p == pathRC && m == "PATCH":
|
||||
if firstPatch {
|
||||
if !isSMPatchVersion_1_5(t, req) {
|
||||
t.Fatalf("apply didn't try to send SMPatchVersion_1_5 for the first time")
|
||||
}
|
||||
firstPatch = false
|
||||
statusErr := kubeerr.NewInternalError(fmt.Errorf("Server encountered internal error."))
|
||||
bodyBytes, _ := json.Marshal(statusErr)
|
||||
bodyErr := ioutil.NopCloser(bytes.NewReader(bodyBytes))
|
||||
return &http.Response{StatusCode: http.StatusInternalServerError, Header: defaultHeader(), Body: bodyErr}, nil
|
||||
}
|
||||
retry = true
|
||||
if isSMPatchVersion_1_5(t, req) {
|
||||
t.Fatalf("apply didn't try to send SMPatchVersion_1_0 after SMPatchVersion_1_5 patch encounter an Internal Error (500)")
|
||||
}
|
||||
bodyRC := ioutil.NopCloser(bytes.NewReader(currentRC))
|
||||
return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: bodyRC}, nil
|
||||
default:
|
||||
t.Fatalf("unexpected request: %#v\n%#v", req.URL, req)
|
||||
return nil, nil
|
||||
}
|
||||
}),
|
||||
}
|
||||
tf.Namespace = "test"
|
||||
tf.ClientConfig = defaultClientConfig()
|
||||
buf := bytes.NewBuffer([]byte{})
|
||||
|
||||
cmd := NewCmdApply(f, buf)
|
||||
cmd.Flags().Set("filename", filenameRC)
|
||||
cmd.Flags().Set("output", "name")
|
||||
cmd.Run(cmd, []string{})
|
||||
|
||||
if !retry {
|
||||
t.Fatalf("apply didn't retry when get Internal Error (500)")
|
||||
}
|
||||
|
||||
// uses the name from the file, not the response
|
||||
expectRC := "replicationcontroller/" + nameRC + "\n"
|
||||
if buf.String() != expectRC {
|
||||
t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expectRC)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyRetry(t *testing.T) {
|
||||
initTestErrorHandler(t)
|
||||
nameRC, currentRC := readAndAnnotateReplicationController(t, filenameRC)
|
||||
|
@ -424,13 +424,8 @@ func visitToPatch(originalObj runtime.Object, updates *resource.Info,
|
||||
results *editResults,
|
||||
file string) error {
|
||||
|
||||
smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
patchVisitor := resource.NewFlattenListVisitor(updates, resourceMapper)
|
||||
err = patchVisitor.Visit(func(info *resource.Info, incomingErr error) error {
|
||||
err := patchVisitor.Visit(func(info *resource.Info, incomingErr error) error {
|
||||
currOriginalObj := originalObj
|
||||
|
||||
// if we're editing a list, then navigate the list to find the item that we're currently trying to edit
|
||||
@ -491,7 +486,8 @@ func visitToPatch(originalObj runtime.Object, updates *resource.Info,
|
||||
|
||||
preconditions := []strategicpatch.PreconditionFunc{strategicpatch.RequireKeyUnchanged("apiVersion"),
|
||||
strategicpatch.RequireKeyUnchanged("kind"), strategicpatch.RequireMetadataKeyUnchanged("name")}
|
||||
patch, err := strategicpatch.CreateTwoWayMergePatch(originalJS, editedJS, currOriginalObj, smPatchVersion, preconditions...)
|
||||
patch, err := strategicpatch.CreateTwoWayMergePatch(originalJS, editedJS, currOriginalObj, strategicpatch.SMPatchVersion_1_5, preconditions...)
|
||||
// If creating a patch fails, retrying with SMPatchVersion_1_0 is not helpful. So we return the error.
|
||||
if err != nil {
|
||||
glog.V(4).Infof("Unable to calculate diff, no merge is possible: %v", err)
|
||||
if strategicpatch.IsPreconditionFailed(err) {
|
||||
@ -503,8 +499,20 @@ func visitToPatch(originalObj runtime.Object, updates *resource.Info,
|
||||
results.version = defaultVersion
|
||||
patched, err := resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch)
|
||||
if err != nil {
|
||||
fmt.Fprintln(out, results.addError(err, info))
|
||||
return nil
|
||||
// Retry SMPatchVersion_1_0 when applying the SMPatchVersion_1_5 patch returns an Internal Error (500).
|
||||
// Because the failure may be due to the server not supporting the SMPatchVersion_1_5 patch.
|
||||
if errors.IsInternalError(err) {
|
||||
patch, err = strategicpatch.CreateTwoWayMergePatch(originalJS, editedJS, currOriginalObj, strategicpatch.SMPatchVersion_1_0)
|
||||
if err != nil {
|
||||
glog.V(4).Infof("Unable to calculate diff, no merge is possible: %v", err)
|
||||
return err
|
||||
}
|
||||
patched, err = resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch)
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Fprintln(out, results.addError(err, info))
|
||||
return nil
|
||||
}
|
||||
}
|
||||
info.Refresh(patched, true)
|
||||
cmdutil.PrintSuccess(mapper, false, out, info.Mapping.Resource, info.Name, false, "edited")
|
||||
|
@ -192,14 +192,6 @@ func (o *LabelOptions) RunLabel(f cmdutil.Factory, cmd *cobra.Command) error {
|
||||
return err
|
||||
}
|
||||
|
||||
smPatchVersion := strategicpatch.SMPatchVersionLatest
|
||||
if !o.local {
|
||||
smPatchVersion, err = cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// only apply resource version locking on a single resource
|
||||
if !one && len(o.resourceVersion) > 0 {
|
||||
return fmt.Errorf("--resource-version may only be used with a single resource")
|
||||
@ -254,7 +246,8 @@ func (o *LabelOptions) RunLabel(f cmdutil.Factory, cmd *cobra.Command) error {
|
||||
if !reflect.DeepEqual(oldData, newData) {
|
||||
dataChangeMsg = "labeled"
|
||||
}
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, smPatchVersion)
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since we only update labels and change cause, and none of them has list of primitives
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, strategicpatch.SMPatchVersion_1_5)
|
||||
createdPatch := err == nil
|
||||
if err != nil {
|
||||
glog.V(2).Infof("couldn't compute patch: %v", err)
|
||||
|
@ -154,14 +154,6 @@ func RunPatch(f cmdutil.Factory, out io.Writer, cmd *cobra.Command, args []strin
|
||||
return err
|
||||
}
|
||||
|
||||
smPatchVersion := strategicpatch.SMPatchVersionLatest
|
||||
if !options.Local {
|
||||
smPatchVersion, err = cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
count := 0
|
||||
err = r.Visit(func(info *resource.Info, err error) error {
|
||||
if err != nil {
|
||||
@ -185,7 +177,7 @@ func RunPatch(f cmdutil.Factory, out io.Writer, cmd *cobra.Command, args []strin
|
||||
// don't bother checking for failures of this replace, because a failure to indicate the hint doesn't fail the command
|
||||
// also, don't force the replacement. If the replacement fails on a resourceVersion conflict, then it means this
|
||||
// record hint is likely to be invalid anyway, so avoid the bad hint
|
||||
patch, err := cmdutil.ChangeResourcePatch(info, f.Command(), smPatchVersion)
|
||||
patch, err := cmdutil.ChangeResourcePatch(info, f.Command())
|
||||
if err == nil {
|
||||
helper.Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch)
|
||||
}
|
||||
|
@ -32,6 +32,7 @@ go_library(
|
||||
"//pkg/runtime:go_default_library",
|
||||
"//pkg/util/errors:go_default_library",
|
||||
"//pkg/util/interrupt:go_default_library",
|
||||
"//pkg/util/strategicpatch:go_default_library",
|
||||
"//pkg/watch:go_default_library",
|
||||
"//vendor:github.com/renstrom/dedent",
|
||||
"//vendor:github.com/spf13/cobra",
|
||||
|
@ -31,6 +31,7 @@ import (
|
||||
"k8s.io/kubernetes/pkg/kubectl/resource"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
utilerrors "k8s.io/kubernetes/pkg/util/errors"
|
||||
"k8s.io/kubernetes/pkg/util/strategicpatch"
|
||||
)
|
||||
|
||||
// PauseConfig is the start of the data required to perform the operation. As new fields are added, add them here instead of
|
||||
@ -134,7 +135,8 @@ func (o *PauseConfig) CompletePause(f cmdutil.Factory, cmd *cobra.Command, out i
|
||||
|
||||
func (o PauseConfig) RunPause() error {
|
||||
allErrs := []error{}
|
||||
for _, patch := range set.CalculatePatches(o.f, o.Infos, o.Encoder, false, o.Pauser) {
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since Pauser only update a boolean variable
|
||||
for _, patch := range set.CalculatePatches(o.f, o.Infos, o.Encoder, strategicpatch.SMPatchVersion_1_5, o.Pauser) {
|
||||
info := patch.Info
|
||||
if patch.Err != nil {
|
||||
allErrs = append(allErrs, fmt.Errorf("error: %s %q %v", info.Mapping.Resource, info.Name, patch.Err))
|
||||
|
@ -31,6 +31,7 @@ import (
|
||||
"k8s.io/kubernetes/pkg/kubectl/resource"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
utilerrors "k8s.io/kubernetes/pkg/util/errors"
|
||||
"k8s.io/kubernetes/pkg/util/strategicpatch"
|
||||
)
|
||||
|
||||
// ResumeConfig is the start of the data required to perform the operation. As new fields are added, add them here instead of
|
||||
@ -138,7 +139,8 @@ func (o *ResumeConfig) CompleteResume(f cmdutil.Factory, cmd *cobra.Command, out
|
||||
|
||||
func (o ResumeConfig) RunResume() error {
|
||||
allErrs := []error{}
|
||||
for _, patch := range set.CalculatePatches(o.f, o.Infos, o.Encoder, false, o.Resumer) {
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since Resumer only update a boolean variable
|
||||
for _, patch := range set.CalculatePatches(o.f, o.Infos, o.Encoder, strategicpatch.SMPatchVersion_1_5, o.Resumer) {
|
||||
info := patch.Info
|
||||
|
||||
if patch.Err != nil {
|
||||
|
@ -139,11 +139,6 @@ func RunScale(f cmdutil.Factory, out io.Writer, cmd *cobra.Command, args []strin
|
||||
return fmt.Errorf("cannot use --resource-version with multiple resources")
|
||||
}
|
||||
|
||||
smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
counter := 0
|
||||
err = r.Visit(func(info *resource.Info, err error) error {
|
||||
if err != nil {
|
||||
@ -169,7 +164,7 @@ func RunScale(f cmdutil.Factory, out io.Writer, cmd *cobra.Command, args []strin
|
||||
return err
|
||||
}
|
||||
if cmdutil.ShouldRecord(cmd, info) {
|
||||
patchBytes, err := cmdutil.ChangeResourcePatch(info, f.Command(), smPatchVersion)
|
||||
patchBytes, err := cmdutil.ChangeResourcePatch(info, f.Command())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -123,17 +123,8 @@ type Patch struct {
|
||||
// If local is true, it will be default to use SMPatchVersionLatest to calculate a patch without contacting the server to
|
||||
// get the server supported SMPatchVersion. If you are using a patch's Patch field generated in local mode, be careful.
|
||||
// If local is false, it will talk to the server to check which StategicMergePatchVersion to use.
|
||||
func CalculatePatches(f cmdutil.Factory, infos []*resource.Info, encoder runtime.Encoder, local bool, mutateFn func(*resource.Info) (bool, error)) []*Patch {
|
||||
func CalculatePatches(f cmdutil.Factory, infos []*resource.Info, encoder runtime.Encoder, smPatchVersion strategicpatch.StrategicMergePatchVersion, mutateFn func(*resource.Info) (bool, error)) []*Patch {
|
||||
var patches []*Patch
|
||||
smPatchVersion := strategicpatch.SMPatchVersionLatest
|
||||
var err error
|
||||
if !local {
|
||||
smPatchVersion, err = cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
|
||||
if err != nil {
|
||||
return patches
|
||||
}
|
||||
}
|
||||
|
||||
for _, info := range infos {
|
||||
patch := &Patch{Info: info}
|
||||
patch.Before, patch.Err = runtime.Encode(encoder, info.Object)
|
||||
|
@ -164,8 +164,8 @@ func (o *ImageOptions) Validate() error {
|
||||
|
||||
func (o *ImageOptions) Run() error {
|
||||
allErrs := []error{}
|
||||
|
||||
patches := CalculatePatches(o.f, o.Infos, o.Encoder, o.Local, func(info *resource.Info) (bool, error) {
|
||||
// Defauting to SMPatchVersion_1_5, since the func passed in doesn't update any lists of primitive
|
||||
patches := CalculatePatches(o.f, o.Infos, o.Encoder, strategicpatch.SMPatchVersion_1_5, func(info *resource.Info) (bool, error) {
|
||||
transformed := false
|
||||
_, err := o.UpdatePodSpecForObject(info.Object, func(spec *api.PodSpec) error {
|
||||
for name, image := range o.ContainerImages {
|
||||
@ -189,14 +189,6 @@ func (o *ImageOptions) Run() error {
|
||||
return transformed, err
|
||||
})
|
||||
|
||||
smPatchVersion := strategicpatch.SMPatchVersionLatest
|
||||
var err error
|
||||
if !o.Local {
|
||||
smPatchVersion, err = cmdutil.GetServerSupportedSMPatchVersionFromFactory(o.f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, patch := range patches {
|
||||
info := patch.Info
|
||||
if patch.Err != nil {
|
||||
@ -223,7 +215,7 @@ func (o *ImageOptions) Run() error {
|
||||
|
||||
// record this change (for rollout history)
|
||||
if o.Record || cmdutil.ContainsChangeCause(info) {
|
||||
if patch, err := cmdutil.ChangeResourcePatch(info, o.ChangeCause, smPatchVersion); err == nil {
|
||||
if patch, err := cmdutil.ChangeResourcePatch(info, o.ChangeCause); err == nil {
|
||||
if obj, err = resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch); err != nil {
|
||||
fmt.Fprintf(o.Err, "WARNING: changes to %s/%s can't be recorded: %v\n", info.Mapping.Resource, info.Name, err)
|
||||
}
|
||||
|
@ -31,6 +31,7 @@ import (
|
||||
"k8s.io/kubernetes/pkg/kubectl/resource"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
utilerrors "k8s.io/kubernetes/pkg/util/errors"
|
||||
"k8s.io/kubernetes/pkg/util/strategicpatch"
|
||||
)
|
||||
|
||||
var (
|
||||
@ -176,7 +177,8 @@ func (o *ResourcesOptions) Validate() error {
|
||||
|
||||
func (o *ResourcesOptions) Run() error {
|
||||
allErrs := []error{}
|
||||
patches := CalculatePatches(o.f, o.Infos, o.Encoder, cmdutil.GetDryRunFlag(o.Cmd), func(info *resource.Info) (bool, error) {
|
||||
// Defauting to SMPatchVersion_1_5, since the func passed in doesn't update any lists of primitive
|
||||
patches := CalculatePatches(o.f, o.Infos, o.Encoder, strategicpatch.SMPatchVersion_1_5, func(info *resource.Info) (bool, error) {
|
||||
transformed := false
|
||||
_, err := o.UpdatePodSpecForObject(info.Object, func(spec *api.PodSpec) error {
|
||||
containers, _ := selectContainers(spec.Containers, o.ContainerSelector)
|
||||
|
@ -321,11 +321,6 @@ func (o TaintOptions) RunTaint() error {
|
||||
return err
|
||||
}
|
||||
|
||||
smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(o.f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return r.Visit(func(info *resource.Info, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
@ -348,7 +343,8 @@ func (o TaintOptions) RunTaint() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, smPatchVersion)
|
||||
// Defaulting to SMPatchVersion_1_5 is safe, since we don't update list of primitives.
|
||||
patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, strategicpatch.SMPatchVersion_1_5)
|
||||
createdPatch := err == nil
|
||||
if err != nil {
|
||||
glog.V(2).Infof("couldn't compute patch: %v", err)
|
||||
|
@ -427,8 +427,12 @@ func (f *fakeAPIFactory) UnstructuredObject() (meta.RESTMapper, runtime.ObjectTy
|
||||
return cmdutil.NewShortcutExpander(mapper, nil), typer, nil
|
||||
}
|
||||
|
||||
func (f *fakeAPIFactory) Decoder(bool) runtime.Decoder {
|
||||
return testapi.Default.Codec()
|
||||
func (f *fakeAPIFactory) Decoder(toInternal bool) runtime.Decoder {
|
||||
if toInternal {
|
||||
return api.Codecs.UniversalDecoder()
|
||||
} else {
|
||||
return api.Codecs.UniversalDeserializer()
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeAPIFactory) JSONEncoder() runtime.Encoder {
|
||||
|
@ -521,7 +521,7 @@ func RecordChangeCause(obj runtime.Object, changeCause string) error {
|
||||
|
||||
// ChangeResourcePatch creates a strategic merge patch between the origin input resource info
|
||||
// and the annotated with change-cause input resource info.
|
||||
func ChangeResourcePatch(info *resource.Info, changeCause string, smPatchVersion strategicpatch.StrategicMergePatchVersion) ([]byte, error) {
|
||||
func ChangeResourcePatch(info *resource.Info, changeCause string) ([]byte, error) {
|
||||
oldData, err := json.Marshal(info.Object)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -533,7 +533,8 @@ func ChangeResourcePatch(info *resource.Info, changeCause string, smPatchVersion
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return strategicpatch.CreateTwoWayMergePatch(oldData, newData, info.Object, smPatchVersion)
|
||||
// Using SMPatchVersion_1_5, since RecordChangeCause() just update the annotation which is a map[string]string
|
||||
return strategicpatch.CreateTwoWayMergePatch(oldData, newData, info.Object, strategicpatch.SMPatchVersion_1_5)
|
||||
}
|
||||
|
||||
// containsChangeCause checks if input resource info contains change-cause annotation.
|
||||
@ -725,13 +726,3 @@ func RequireNoArguments(c *cobra.Command, args []string) {
|
||||
CheckErr(UsageError(c, fmt.Sprintf(`unknown command %q`, strings.Join(args, " "))))
|
||||
}
|
||||
}
|
||||
|
||||
// GetServerSupportedSMPatchVersionFromFactory is a wrapper of GetServerSupportedSMPatchVersion(),
|
||||
// It takes a Factory, returns the max version the server supports.
|
||||
func GetServerSupportedSMPatchVersionFromFactory(f Factory) (strategicpatch.StrategicMergePatchVersion, error) {
|
||||
clientSet, err := f.ClientSet()
|
||||
if err != nil {
|
||||
return strategicpatch.Unknown, err
|
||||
}
|
||||
return strategicpatch.GetServerSupportedSMPatchVersion(clientSet.Discovery())
|
||||
}
|
||||
|
@ -15,7 +15,6 @@ go_library(
|
||||
srcs = ["patch.go"],
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//pkg/client/typed/discovery:go_default_library",
|
||||
"//pkg/util/json:go_default_library",
|
||||
"//third_party/forked/golang/json:go_default_library",
|
||||
"//vendor:github.com/davecgh/go-spew/spew",
|
||||
|
@ -21,7 +21,6 @@ import (
|
||||
"reflect"
|
||||
"sort"
|
||||
|
||||
"k8s.io/kubernetes/pkg/client/typed/discovery"
|
||||
"k8s.io/kubernetes/pkg/util/json"
|
||||
forkedjson "k8s.io/kubernetes/third_party/forked/golang/json"
|
||||
|
||||
@ -46,7 +45,7 @@ const (
|
||||
deleteDirective = "delete"
|
||||
replaceDirective = "replace"
|
||||
mergeDirective = "merge"
|
||||
mergePrimitivesListDirective = "mergeprimitiveslist"
|
||||
MergePrimitivesListDirective = "mergeprimitiveslist"
|
||||
|
||||
// different versions of StrategicMergePatch
|
||||
SMPatchVersion_1_0 StrategicMergePatchVersion = "v1.0.0"
|
||||
@ -394,7 +393,7 @@ loopB:
|
||||
func diffListsOfScalarsIntoMap(originalScalars, modifiedScalars []interface{}, ignoreChangesAndAdditions, ignoreDeletions bool) (map[string]interface{}, error) {
|
||||
originalIndex, modifiedIndex := 0, 0
|
||||
patch := map[string]interface{}{}
|
||||
patch[directiveMarker] = mergePrimitivesListDirective
|
||||
patch[directiveMarker] = MergePrimitivesListDirective
|
||||
|
||||
for originalIndex < len(originalScalars) && modifiedIndex < len(modifiedScalars) {
|
||||
originalString := fmt.Sprintf("%v", originalScalars[originalIndex])
|
||||
@ -628,7 +627,7 @@ func mergeMap(original, patch map[string]interface{}, t reflect.Type) (map[strin
|
||||
return map[string]interface{}{}, nil
|
||||
}
|
||||
|
||||
if v == mergePrimitivesListDirective {
|
||||
if v == MergePrimitivesListDirective {
|
||||
// delete the directiveMarker's key-value pair to avoid delta map and delete map
|
||||
// overlaping with each other when calculating a ThreeWayDiff for list of Primitives.
|
||||
// Otherwise, the overlaping will cause it calling LookupPatchMetadata() which will
|
||||
@ -719,7 +718,7 @@ func mergeMap(original, patch map[string]interface{}, t reflect.Type) (map[strin
|
||||
// the patch because getting a deep copy of a slice in golang is highly
|
||||
// non-trivial.
|
||||
// The patch could be a map[string]interface{} representing a slice of primitives.
|
||||
// If the patch map doesn't has the specific directiveMarker (mergePrimitivesListDirective),
|
||||
// If the patch map doesn't has the specific directiveMarker (MergePrimitivesListDirective),
|
||||
// it returns an error. Please check patch_test.go and find the test case named
|
||||
// "merge lists of scalars for list of primitives" to see what the patch looks like.
|
||||
// Patch is still []interface{} for all the other types.
|
||||
@ -732,7 +731,7 @@ func mergeSlice(original []interface{}, patch interface{}, elemType reflect.Type
|
||||
if patchMap, ok := patch.(map[string]interface{}); ok {
|
||||
// We try to merge the original slice with a patch map only when the map has
|
||||
// a specific directiveMarker. Otherwise, this patch will be treated as invalid.
|
||||
if directiveValue, ok := patchMap[directiveMarker]; ok && directiveValue == mergePrimitivesListDirective {
|
||||
if directiveValue, ok := patchMap[directiveMarker]; ok && directiveValue == MergePrimitivesListDirective {
|
||||
return mergeSliceOfScalarsWithPatchMap(original, patchMap)
|
||||
} else {
|
||||
return nil, fmt.Errorf("Unable to merge a slice with an invalid map")
|
||||
@ -839,10 +838,10 @@ func mergeSlice(original []interface{}, patch interface{}, elemType reflect.Type
|
||||
|
||||
// mergeSliceOfScalarsWithPatchMap merges the original slice with a patch map and
|
||||
// returns an uniqified and sorted slice of primitives.
|
||||
// The patch map must have the specific directiveMarker (mergePrimitivesListDirective).
|
||||
// The patch map must have the specific directiveMarker (MergePrimitivesListDirective).
|
||||
func mergeSliceOfScalarsWithPatchMap(original []interface{}, patch map[string]interface{}) ([]interface{}, error) {
|
||||
// make sure the patch has the specific directiveMarker ()
|
||||
if directiveValue, ok := patch[directiveMarker]; ok && directiveValue != mergePrimitivesListDirective {
|
||||
if directiveValue, ok := patch[directiveMarker]; ok && directiveValue != MergePrimitivesListDirective {
|
||||
return nil, fmt.Errorf("Unable to merge a slice with an invalid map")
|
||||
}
|
||||
delete(patch, directiveMarker)
|
||||
@ -1182,7 +1181,7 @@ func mergingMapFieldsHaveConflicts(
|
||||
return true, nil
|
||||
}
|
||||
|
||||
if leftMarker == mergePrimitivesListDirective && rightMarker == mergePrimitivesListDirective {
|
||||
if leftMarker == MergePrimitivesListDirective && rightMarker == MergePrimitivesListDirective {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
@ -1210,7 +1209,7 @@ func mapsHaveConflicts(typedLeft, typedRight map[string]interface{}, structType
|
||||
isForListOfPrimitives := false
|
||||
if leftDirective, ok := typedLeft[directiveMarker]; ok {
|
||||
if rightDirective, ok := typedRight[directiveMarker]; ok {
|
||||
if leftDirective == mergePrimitivesListDirective && rightDirective == rightDirective {
|
||||
if leftDirective == MergePrimitivesListDirective && rightDirective == rightDirective {
|
||||
isForListOfPrimitives = true
|
||||
}
|
||||
}
|
||||
@ -1430,20 +1429,3 @@ func toYAML(v interface{}) (string, error) {
|
||||
|
||||
return string(y), nil
|
||||
}
|
||||
|
||||
// GetServerSupportedSMPatchVersion takes a discoveryClient,
|
||||
// returns the max StrategicMergePatch version supported
|
||||
func GetServerSupportedSMPatchVersion(discoveryClient discovery.DiscoveryInterface) (StrategicMergePatchVersion, error) {
|
||||
serverVersion, err := discoveryClient.ServerVersion()
|
||||
if err != nil {
|
||||
return Unknown, err
|
||||
}
|
||||
serverGitVersion := serverVersion.GitVersion
|
||||
if serverGitVersion >= string(SMPatchVersion_1_5) {
|
||||
return SMPatchVersion_1_5, nil
|
||||
}
|
||||
if serverGitVersion >= string(SMPatchVersion_1_0) {
|
||||
return SMPatchVersion_1_0, nil
|
||||
}
|
||||
return Unknown, fmt.Errorf("The version is too old: %v\n", serverVersion)
|
||||
}
|
||||
|
2
test/fixtures/pkg/kubectl/cmd/apply/rc.yaml
vendored
2
test/fixtures/pkg/kubectl/cmd/apply/rc.yaml
vendored
@ -1,6 +1,8 @@
|
||||
apiVersion: v1
|
||||
kind: ReplicationController
|
||||
metadata:
|
||||
finalizers:
|
||||
- b/b
|
||||
name: test-rc
|
||||
labels:
|
||||
name: test-rc
|
||||
|
Loading…
Reference in New Issue
Block a user