// Copyright 2018 Google Inc. All Rights Reserved. // // Licensed 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 fleetallocation import ( "encoding/json" "fmt" "sync" "agones.dev/agones/pkg/apis/stable" stablev1alpha1 "agones.dev/agones/pkg/apis/stable/v1alpha1" "agones.dev/agones/pkg/client/clientset/versioned" getterv1alpha1 "agones.dev/agones/pkg/client/clientset/versioned/typed/stable/v1alpha1" "agones.dev/agones/pkg/client/informers/externalversions" listerv1alpha1 "agones.dev/agones/pkg/client/listers/stable/v1alpha1" "agones.dev/agones/pkg/fleets" "agones.dev/agones/pkg/util/crd" "agones.dev/agones/pkg/util/runtime" "agones.dev/agones/pkg/util/webhooks" "github.com/mattbaird/jsonpatch" "github.com/pkg/errors" "github.com/sirupsen/logrus" admv1beta1 "k8s.io/api/admission/v1beta1" corev1 "k8s.io/api/core/v1" extclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1" k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/record" ) var ( // ErrNoGameServerReady is returned when there are no Ready GameServers // available ErrNoGameServerReady = errors.New("Could not find a Ready GameServer") ) // Controller is a the FleetAllocation controller type Controller struct { logger *logrus.Entry crdGetter v1beta1.CustomResourceDefinitionInterface gameServerSynced cache.InformerSynced gameServerGetter getterv1alpha1.GameServersGetter gameServerLister listerv1alpha1.GameServerLister gameServerSetLister listerv1alpha1.GameServerSetLister fleetLister listerv1alpha1.FleetLister fleetAllocationGetter getterv1alpha1.FleetAllocationsGetter fleetAllocationLister listerv1alpha1.FleetAllocationLister stop <-chan struct{} allocationMutex *sync.Mutex recorder record.EventRecorder } // NewController returns a controller for a FleetAllocation func NewController( wh *webhooks.WebHook, allocationMutex *sync.Mutex, kubeClient kubernetes.Interface, extClient extclientset.Interface, agonesClient versioned.Interface, agonesInformerFactory externalversions.SharedInformerFactory) *Controller { agonesInformer := agonesInformerFactory.Stable().V1alpha1() c := &Controller{ crdGetter: extClient.ApiextensionsV1beta1().CustomResourceDefinitions(), gameServerSynced: agonesInformer.GameServers().Informer().HasSynced, gameServerGetter: agonesClient.StableV1alpha1(), gameServerLister: agonesInformer.GameServers().Lister(), gameServerSetLister: agonesInformer.GameServerSets().Lister(), fleetLister: agonesInformer.Fleets().Lister(), fleetAllocationGetter: agonesClient.StableV1alpha1(), fleetAllocationLister: agonesInformer.FleetAllocations().Lister(), allocationMutex: allocationMutex, } c.logger = runtime.NewLoggerWithType(c) eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(c.logger.Infof) eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")}) c.recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "fleetallocation-controller"}) kind := stablev1alpha1.Kind("FleetAllocation") wh.AddHandler("/mutate", kind, admv1beta1.Create, c.creationMutationHandler) wh.AddHandler("/validate", kind, admv1beta1.Create, c.creationValidationHandler) wh.AddHandler("/validate", kind, admv1beta1.Update, c.mutationValidationHandler) return c } // Run runs this controller. This controller doesn't (currently) // have a worker/queue, and as such, does not block. func (c *Controller) Run(workers int, stop <-chan struct{}) error { err := crd.WaitForEstablishedCRD(c.crdGetter, "fleetallocations."+stable.GroupName, c.logger) if err != nil { return err } c.stop = stop return nil } // creationMutationHandler will intercept when a FleetAllocation is created, and allocate it a GameServer // assuming that one is available. If not, it will reject the AdmissionReview. func (c *Controller) creationMutationHandler(review admv1beta1.AdmissionReview) (admv1beta1.AdmissionReview, error) { c.logger.WithField("review", review).Info("creationMutationHandler") obj := review.Request.Object fa := &stablev1alpha1.FleetAllocation{} err := json.Unmarshal(obj.Raw, fa) if err != nil { return review, errors.Wrapf(err, "error unmarshalling original FleetAllocation json: %s", obj.Raw) } // When being called from the API the fa.ObjectMeta.Namespace isn't populated // (whereas it is from kubectl). So make sure to pull the namespace from the review fleet, err := c.fleetLister.Fleets(review.Request.Namespace).Get(fa.Spec.FleetName) if err != nil { if k8serrors.IsNotFound(err) { logrus.WithError(err).WithField("fleetName", fa.Name). WithField("namespace", review.Request.Namespace). Warn("Could not find fleet for allocation. Skipping.") return review, nil } return review, errors.Wrapf(err, "error retrieving fleet %s", fa.Name) } gs, err := c.allocate(fleet, &fa.Spec.MetaPatch) if err != nil { review.Response.Allowed = false review.Response.Result = &metav1.Status{ Status: metav1.StatusFailure, Reason: metav1.StatusReasonNotFound, Details: &metav1.StatusDetails{ Name: review.Request.Name, Group: review.Request.Kind.Group, Kind: "GameServer", }, } return review, nil } // When a GameServer is deleted, the FleetAllocation should go with it ref := metav1.NewControllerRef(gs, stablev1alpha1.SchemeGroupVersion.WithKind("GameServer")) fa.ObjectMeta.OwnerReferences = append(fa.ObjectMeta.OwnerReferences, *ref) fa.Status = stablev1alpha1.FleetAllocationStatus{GameServer: gs} newFA, err := json.Marshal(fa) if err != nil { return review, errors.Wrapf(err, "error marshalling FleetAllocation %s to json", fa.ObjectMeta.Name) } patch, err := jsonpatch.CreatePatch(obj.Raw, newFA) if err != nil { return review, errors.Wrapf(err, "error creating patch for FleetAllocation %s", fa.ObjectMeta.Name) } json, err := json.Marshal(patch) if err != nil { return review, errors.Wrapf(err, "error creating json for patch for FleetAllocation %s", gs.ObjectMeta.Name) } c.logger.WithField("fa", fa.ObjectMeta.Name).WithField("patch", string(json)).Infof("patch created!") pt := admv1beta1.PatchTypeJSONPatch review.Response.PatchType = &pt review.Response.Patch = json return review, nil } // creationValidationHandler intercepts the creation of a FleetAllocation, and if there is // no Status > GameServer set, then we will assume that the Spec > fleetName is invalid func (c *Controller) creationValidationHandler(review admv1beta1.AdmissionReview) (admv1beta1.AdmissionReview, error) { c.logger.WithField("review", review).Info("creationValidationHandler") obj := review.Request.Object fa := &stablev1alpha1.FleetAllocation{} if err := json.Unmarshal(obj.Raw, fa); err != nil { return review, errors.Wrapf(err, "error unmarshalling original FleetAllocation json: %s", obj.Raw) } // If there is no GameServer, we are assuming that is // because the fleetName is invalid. Any other error // option should be handled by the creationMutationHandler if fa.Status.GameServer == nil { review.Response.Allowed = false review.Response.Result = &metav1.Status{ Status: metav1.StatusFailure, Reason: metav1.StatusReasonInvalid, Message: "Invalid FleetAllocation", Details: &metav1.StatusDetails{ Name: review.Request.Name, Group: review.Request.Kind.Group, Kind: review.Request.Kind.Kind, Causes: []metav1.StatusCause{ {Type: metav1.CauseTypeFieldValueNotFound, Message: fmt.Sprintf("Could not find fleet %s in namespace %s", fa.Spec.FleetName, review.Request.Namespace), Field: "fleetName"}}, }, } } return review, nil } // mutationValidationHandler stops edits from happening to a // FleetAllocation fleetName value func (c *Controller) mutationValidationHandler(review admv1beta1.AdmissionReview) (admv1beta1.AdmissionReview, error) { c.logger.WithField("review", review).Info("mutationValidationHandler") newFA := &stablev1alpha1.FleetAllocation{} oldFA := &stablev1alpha1.FleetAllocation{} if err := json.Unmarshal(review.Request.Object.Raw, newFA); err != nil { return review, errors.Wrapf(err, "error unmarshalling new FleetAllocation json: %s", review.Request.Object.Raw) } if err := json.Unmarshal(review.Request.OldObject.Raw, oldFA); err != nil { return review, errors.Wrapf(err, "error unmarshalling old FleetAllocation json: %s", review.Request.Object.Raw) } if ok, causes := oldFA.ValidateUpdate(newFA); !ok { review.Response.Allowed = false details := metav1.StatusDetails{ Name: review.Request.Name, Group: review.Request.Kind.Group, Kind: review.Request.Kind.Kind, Causes: causes, } review.Response.Result = &metav1.Status{ Status: metav1.StatusFailure, Message: "FleetAllocation update is invalid", Reason: metav1.StatusReasonInvalid, Details: &details, } } return review, nil } // allocate allocated a GameServer from a given Fleet func (c *Controller) allocate(f *stablev1alpha1.Fleet, fam *stablev1alpha1.FleetAllocationMeta) (*stablev1alpha1.GameServer, error) { var allocation *stablev1alpha1.GameServer // can only allocate one at a time, as we don't want two separate processes // trying to allocate the same GameServer to different clients c.allocationMutex.Lock() defer c.allocationMutex.Unlock() // make sure we have the most up to date view of the world if !cache.WaitForCacheSync(c.stop, c.gameServerSynced) { return allocation, errors.New("error syncing GameServer cache") } gsList, err := fleets.ListGameServersByFleetOwner(c.gameServerLister, c.gameServerSetLister, f) if err != nil { return allocation, err } for _, gs := range gsList { if gs.Status.State == stablev1alpha1.Ready && gs.ObjectMeta.DeletionTimestamp.IsZero() { allocation = gs break } } if allocation == nil { return allocation, ErrNoGameServerReady } gsCopy := allocation.DeepCopy() gsCopy.Status.State = stablev1alpha1.Allocated if fam != nil { c.patchMetadata(gsCopy, fam) } gs, err := c.gameServerGetter.GameServers(f.ObjectMeta.Namespace).Update(gsCopy) if err != nil { return gs, errors.Wrapf(err, "error updating GameServer %s", gsCopy.ObjectMeta.Name) } c.recorder.Eventf(gs, corev1.EventTypeNormal, string(gs.Status.State), "Allocated from Fleet %s", f.ObjectMeta.Name) return gs, nil } // patch the labels and annotations of an allocated GameServer with metadata from a FleetAllocation func (c *Controller) patchMetadata(gs *stablev1alpha1.GameServer, fam *stablev1alpha1.FleetAllocationMeta) { // patch ObjectMeta labels if fam.Labels != nil { if gs.ObjectMeta.Labels == nil { gs.ObjectMeta.Labels = make(map[string]string, len(fam.Labels)) } for key, value := range fam.Labels { gs.ObjectMeta.Labels[key] = value } } // apply annotations patch if fam.Annotations != nil { if gs.ObjectMeta.Annotations == nil { gs.ObjectMeta.Annotations = make(map[string]string, len(fam.Annotations)) } for key, value := range fam.Annotations { gs.ObjectMeta.Annotations[key] = value } } }