Skip to content

Commit

Permalink
Fix apache#1107: generalize server side apply code and reuse
Browse files Browse the repository at this point in the history
  • Loading branch information
nicolaferraro committed Dec 20, 2021
1 parent 32b0898 commit ca8509c
Show file tree
Hide file tree
Showing 5 changed files with 136 additions and 87 deletions.
6 changes: 3 additions & 3 deletions addons/keda/keda.go
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ func (t *kedaTrait) getScaledObject(e *trait.Environment) (*kedav1alpha1.ScaledO

func (t *kedaTrait) hackControllerReplicas(e *trait.Environment) error {
ctrlRef := t.getTopControllerReference(e)

applier := e.Client.ServerOrClientSideApplier()
if ctrlRef.Kind == camelv1alpha1.KameletBindingKind {
// Update the KameletBinding directly (do not add it to env resources, it's the integration parent)
key := client.ObjectKey{
Expand All @@ -131,15 +131,15 @@ func (t *kedaTrait) hackControllerReplicas(e *trait.Environment) error {
if klb.Spec.Replicas == nil {
one := int32(1)
klb.Spec.Replicas = &one
if err := e.Client.Update(e.Ctx, &klb); err != nil {
if err := applier.Apply(e.Ctx, &klb); err != nil {
return err
}
}
} else {
if e.Integration.Spec.Replicas == nil {
one := int32(1)
e.Integration.Spec.Replicas = &one
if err := e.Client.Update(e.Ctx, e.Integration); err != nil {
if err := applier.Apply(e.Ctx, e.Integration); err != nil {
return err
}
}
Expand Down
1 change: 1 addition & 0 deletions pkg/client/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ type Client interface {
GetScheme() *runtime.Scheme
GetConfig() *rest.Config
GetCurrentNamespace(kubeConfig string) (string, error)
ServerOrClientSideApplier() ServerOrClientSideApplier
}

// Injectable identifies objects that can receive a Client.
Expand Down
124 changes: 124 additions & 0 deletions pkg/client/serverside.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
/*
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package client

import (
"context"
"fmt"
"net/http"
"strings"
"sync"
"sync/atomic"

"github.com/apache/camel-k/pkg/util/log"
"github.com/apache/camel-k/pkg/util/patch"
"github.com/pkg/errors"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/types"
ctrl "sigs.k8s.io/controller-runtime/pkg/client"
)

type ServerOrClientSideApplier struct {
Client ctrl.Client
hasServerSideApply atomic.Value
tryServerSideApply sync.Once
}

func (c *defaultClient) ServerOrClientSideApplier() ServerOrClientSideApplier {
return ServerOrClientSideApplier{
Client: c,
}
}

func (a *ServerOrClientSideApplier) Apply(ctx context.Context, object ctrl.Object) error {
once := false
var err error
a.tryServerSideApply.Do(func() {
once = true
if err = a.serverSideApply(ctx, object); err != nil {
if isIncompatibleServerError(err) {
log.Info("Fallback to client-side apply for installing resources")
a.hasServerSideApply.Store(false)
err = nil
} else {
a.tryServerSideApply = sync.Once{}
}
} else {
a.hasServerSideApply.Store(true)
}
})
if err != nil {
return err
}
if v := a.hasServerSideApply.Load(); v.(bool) {
if !once {
return a.serverSideApply(ctx, object)
}
} else {
return a.clientSideApply(ctx, object)
}
return nil
}

func (a *ServerOrClientSideApplier) serverSideApply(ctx context.Context, resource ctrl.Object) error {
target, err := patch.PositiveApplyPatch(resource)
if err != nil {
return err
}
return a.Client.Patch(ctx, target, ctrl.Apply, ctrl.ForceOwnership, ctrl.FieldOwner("camel-k-operator"))
}

func (a *ServerOrClientSideApplier) clientSideApply(ctx context.Context, resource ctrl.Object) error {
err := a.Client.Create(ctx, resource)
if err == nil {
return nil
} else if !k8serrors.IsAlreadyExists(err) {
return fmt.Errorf("error during create resource: %s/%s: %w", resource.GetNamespace(), resource.GetName(), err)
}
object := &unstructured.Unstructured{}
object.SetNamespace(resource.GetNamespace())
object.SetName(resource.GetName())
object.SetGroupVersionKind(resource.GetObjectKind().GroupVersionKind())
err = a.Client.Get(ctx, ctrl.ObjectKeyFromObject(object), object)
if err != nil {
return err
}
p, err := patch.PositiveMergePatch(object, resource)
if err != nil {
return err
} else if len(p) == 0 {
return nil
}
return a.Client.Patch(ctx, resource, ctrl.RawPatch(types.MergePatchType, p))
}

func isIncompatibleServerError(err error) bool {
// First simpler check for older servers (i.e. OpenShift 3.11)
if strings.Contains(err.Error(), "415: Unsupported Media Type") {
return true
}
// 415: Unsupported media type means we're talking to a server which doesn't
// support server-side apply.
var serr *k8serrors.StatusError
if errors.As(err, &serr) {
return serr.Status().Code == http.StatusUnsupportedMediaType
}
// Non-StatusError means the error isn't because the server is incompatible.
return false
}
86 changes: 2 additions & 84 deletions pkg/install/kamelets.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,33 +19,23 @@ package install

import (
"context"
"errors"
"fmt"
"io/fs"
"net/http"
"os"
"path"
"path/filepath"
"strings"
"sync"
"sync/atomic"

"golang.org/x/sync/errgroup"

k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"

ctrl "sigs.k8s.io/controller-runtime/pkg/client"
logf "sigs.k8s.io/controller-runtime/pkg/log"

"github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
"github.com/apache/camel-k/pkg/client"
"github.com/apache/camel-k/pkg/util"
"github.com/apache/camel-k/pkg/util/defaults"
"github.com/apache/camel-k/pkg/util/kubernetes"
"github.com/apache/camel-k/pkg/util/patch"
)

const (
Expand All @@ -55,9 +45,6 @@ const (

var (
log = logf.Log

hasServerSideApply atomic.Value
tryServerSideApply sync.Once
)

// KameletCatalog installs the bundled Kamelets into the specified namespace.
Expand All @@ -77,7 +64,7 @@ func KameletCatalog(ctx context.Context, c client.Client, namespace string) erro
}

g, gCtx := errgroup.WithContext(ctx)

applier := c.ServerOrClientSideApplier()
err = filepath.WalkDir(kameletDir, func(p string, f fs.DirEntry, err error) error {
if err != nil {
return err
Expand All @@ -94,31 +81,9 @@ func KameletCatalog(ctx context.Context, c client.Client, namespace string) erro
if err != nil {
return err
}
once := false
tryServerSideApply.Do(func() {
once = true
if err = serverSideApply(gCtx, c, kamelet); err != nil {
if isIncompatibleServerError(err) {
log.Info("Fallback to client-side apply for installing bundled Kamelets")
hasServerSideApply.Store(false)
err = nil
} else {
tryServerSideApply = sync.Once{}
}
} else {
hasServerSideApply.Store(true)
}
})
if err != nil {
if err := applier.Apply(gCtx, kamelet); err != nil {
return err
}
if v := hasServerSideApply.Load(); v.(bool) {
if !once {
return serverSideApply(gCtx, c, kamelet)
}
} else {
return clientSideApply(gCtx, c, kamelet)
}
return nil
})
return nil
Expand All @@ -130,53 +95,6 @@ func KameletCatalog(ctx context.Context, c client.Client, namespace string) erro
return g.Wait()
}

func serverSideApply(ctx context.Context, c client.Client, resource runtime.Object) error {
target, err := patch.PositiveApplyPatch(resource)
if err != nil {
return err
}
return c.Patch(ctx, target, ctrl.Apply, ctrl.ForceOwnership, ctrl.FieldOwner("camel-k-operator"))
}

func clientSideApply(ctx context.Context, c client.Client, resource ctrl.Object) error {
err := c.Create(ctx, resource)
if err == nil {
return nil
} else if !k8serrors.IsAlreadyExists(err) {
return fmt.Errorf("error during create resource: %s/%s: %w", resource.GetNamespace(), resource.GetName(), err)
}
object := &unstructured.Unstructured{}
object.SetNamespace(resource.GetNamespace())
object.SetName(resource.GetName())
object.SetGroupVersionKind(resource.GetObjectKind().GroupVersionKind())
err = c.Get(ctx, ctrl.ObjectKeyFromObject(object), object)
if err != nil {
return err
}
p, err := patch.PositiveMergePatch(object, resource)
if err != nil {
return err
} else if len(p) == 0 {
return nil
}
return c.Patch(ctx, resource, ctrl.RawPatch(types.MergePatchType, p))
}

func isIncompatibleServerError(err error) bool {
// First simpler check for older servers (i.e. OpenShift 3.11)
if strings.Contains(err.Error(), "415: Unsupported Media Type") {
return true
}
// 415: Unsupported media type means we're talking to a server which doesn't
// support server-side apply.
var serr *k8serrors.StatusError
if errors.As(err, &serr) {
return serr.Status().Code == http.StatusUnsupportedMediaType
}
// Non-StatusError means the error isn't because the server is incompatible.
return false
}

func loadKamelet(path string, namespace string, scheme *runtime.Scheme) (*v1alpha1.Kamelet, error) {
content, err := util.ReadFile(path)
if err != nil {
Expand Down
6 changes: 6 additions & 0 deletions pkg/util/test/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,12 @@ func (c *FakeClient) Discovery() discovery.DiscoveryInterface {
}
}

func (c *FakeClient) ServerOrClientSideApplier() client.ServerOrClientSideApplier {
return client.ServerOrClientSideApplier{
Client: c,
}
}

type FakeDiscovery struct {
discovery.DiscoveryInterface
}
Expand Down

0 comments on commit ca8509c

Please sign in to comment.