diff --git a/pkg/gameserverallocations/allocation_cache.go b/pkg/gameserverallocations/allocation_cache.go index d01af7ac7b..1ffdfec649 100644 --- a/pkg/gameserverallocations/allocation_cache.go +++ b/pkg/gameserverallocations/allocation_cache.go @@ -24,11 +24,11 @@ import ( informerv1 "agones.dev/agones/pkg/client/informers/externalversions/agones/v1" listerv1 "agones.dev/agones/pkg/client/listers/agones/v1" "agones.dev/agones/pkg/gameservers" + "agones.dev/agones/pkg/util/errors" "agones.dev/agones/pkg/util/logfields" "agones.dev/agones/pkg/util/runtime" "agones.dev/agones/pkg/util/workerqueue" "github.com/heptiolabs/healthcheck" - "github.com/pkg/errors" "github.com/sirupsen/logrus" "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/tools/cache" @@ -50,6 +50,7 @@ type AllocationCache struct { workerqueue *workerqueue.WorkerQueue counter *gameservers.PerNodeCounter matcher matcher + errs *errors.Errors } // NewAllocationCache creates a new instance of AllocationCache @@ -95,6 +96,7 @@ func NewAllocationCache(informer informerv1.GameServerInformer, counter *gameser }) c.baseLogger = runtime.NewLoggerWithType(c) + c.errs = errors.FromStruct(c) c.workerqueue = workerqueue.NewWorkerQueue(c.SyncGameServers, c.baseLogger, logfields.GameServerKey, agones.GroupName+".AllocationCache") health.AddLivenessCheck("allocationcache-workerqueue", healthcheck.Check(c.workerqueue.Healthy)) @@ -118,7 +120,7 @@ func (c *AllocationCache) RemoveGameServer(gs *agonesv1.GameServer) error { func (c *AllocationCache) Sync(ctx context.Context) error { c.baseLogger.Debug("Wait for AllocationCache cache sync") if !cache.WaitForCacheSync(ctx.Done(), c.gameServerSynced) { - return errors.New("failed to wait for caches to sync") + return c.errs.New("failed to wait for caches to sync") } // build the cache @@ -260,7 +262,7 @@ func (c *AllocationCache) syncCache() error { // build the cache gsList, err := c.gameServerLister.List(labels.Everything()) if err != nil { - return errors.Wrap(err, "could not list GameServers") + return c.errs.Wrap(err, "could not list GameServers") } // convert list of current gameservers to map for faster access @@ -307,7 +309,7 @@ func (c *AllocationCache) getKey(gs *agonesv1.GameServer) (string, bool) { var err error if key, err = cache.MetaNamespaceKeyFunc(gs); err != nil { ok = false - err = errors.Wrap(err, "Error creating key for object") + err = c.errs.Wrap(err, "Error creating key for object") runtime.HandleError(c.baseLogger.WithField("obj", gs), err) } return key, ok diff --git a/pkg/gameserverallocations/allocator.go b/pkg/gameserverallocations/allocator.go index c2a884de2c..b8683670fd 100644 --- a/pkg/gameserverallocations/allocator.go +++ b/pkg/gameserverallocations/allocator.go @@ -33,9 +33,9 @@ import ( multiclusterinformerv1 "agones.dev/agones/pkg/client/informers/externalversions/multicluster/v1" multiclusterlisterv1 "agones.dev/agones/pkg/client/listers/multicluster/v1" "agones.dev/agones/pkg/util/apiserver" + "agones.dev/agones/pkg/util/errors" "agones.dev/agones/pkg/util/logfields" "agones.dev/agones/pkg/util/runtime" - "github.com/pkg/errors" "github.com/sirupsen/logrus" "go.opencensus.io/tag" "google.golang.org/grpc" @@ -61,13 +61,13 @@ import ( var ( // ErrNoGameServer is returned when there are no Allocatable GameServers // available - ErrNoGameServer = errors.New("Could not find an Allocatable GameServer") + ErrNoGameServer = errs.New("Could not find an Allocatable GameServer") // ErrConflictInGameServerSelection is returned when the candidate gameserver already allocated - ErrConflictInGameServerSelection = errors.New("The Gameserver was already allocated") + ErrConflictInGameServerSelection = errs.New("The Gameserver was already allocated") // ErrTotalTimeoutExceeded is used to signal that total retry timeout has been exceeded and no additional retries should be made ErrTotalTimeoutExceeded = status.Errorf(codes.DeadlineExceeded, "remote allocation total timeout exceeded") // ErrGameServerUpdateConflict is returned when the game server selected for applying the allocation cannot be updated - ErrGameServerUpdateConflict = errors.New("could not update the selected GameServer") + ErrGameServerUpdateConflict = errs.New("could not update the selected GameServer") ) const ( @@ -98,6 +98,8 @@ var remoteAllocationRetry = wait.Backoff{ } // Allocator handles game server allocation +// +//nolint:govet // fieldalignment: struct alignment is not critical for our use case type Allocator struct { baseLogger *logrus.Entry allocationPolicyLister multiclusterlisterv1.GameServerAllocationPolicyLister @@ -112,6 +114,7 @@ type Allocator struct { remoteAllocationTimeout time.Duration totalRemoteAllocationTimeout time.Duration batchWaitTime time.Duration + errs *errors.Errors } // request is an async request for allocation @@ -157,6 +160,7 @@ func NewAllocator(policyInformer multiclusterinformerv1.GameServerAllocationPoli } ah.baseLogger = runtime.NewLoggerWithType(ah) + ah.errs = errors.FromStruct(ah) eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(ah.baseLogger.Debugf) eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")}) @@ -185,7 +189,7 @@ func (c *Allocator) Run(ctx context.Context) error { func (c *Allocator) Sync(ctx context.Context) error { c.baseLogger.Debug("Wait for Allocator cache sync") if !cache.WaitForCacheSync(ctx.Done(), c.secretSynced, c.allocationPolicySynced) { - return errors.New("failed to wait for caches to sync") + return c.errs.New("failed to wait for caches to sync") } return nil } @@ -212,7 +216,7 @@ func (c *Allocator) Allocate(ctx context.Context, gsa *allocationv1.GameServerAl var gvks []schema.GroupVersionKind gvks, _, err := apiserver.Scheme.ObjectKinds(s) if err != nil { - return nil, errors.Wrap(err, "could not find objectkinds for status") + return nil, c.errs.Wrap(err, "could not find objectkinds for status") } c.loggerForGameServerAllocation(gsa).Debug("GameServerAllocation is invalid") @@ -318,7 +322,7 @@ func (c *Allocator) applyMultiClusterAllocation(ctx context.Context, gsa *alloca if err != nil { return nil, err } else if len(policies) == 0 { - return nil, errors.New("no multi-cluster allocation policy is specified") + return nil, c.errs.New("no multi-cluster allocation policy is specified") } it := multiclusterv1.NewConnectionInfoIterator(policies) @@ -426,7 +430,7 @@ func (c *Allocator) createRemoteClusterDialOption(namespace string, connectionIn // This is required for self-signed certs. tlsConfig.RootCAs = x509.NewCertPool() if len(connectionInfo.ServerCA) != 0 && !tlsConfig.RootCAs.AppendCertsFromPEM(connectionInfo.ServerCA) { - return nil, errors.New("only PEM format is accepted for server CA") + return nil, c.errs.New("only PEM format is accepted for server CA") } // Add client CA cert, which can be used instead of / as well as the specified ServerCA cert if len(caCert) != 0 { @@ -608,7 +612,7 @@ func (c *Allocator) allocationUpdateWorkers(ctx context.Context, workerCount int case res := <-updateQueue: gs, err := c.applyAllocationToGameServer(ctx, res.request.gsa.Spec.MetaPatch, res.gs, res.request.gsa) if err != nil { - if !k8serrors.IsConflict(errors.Cause(err)) { + if !k8serrors.IsConflict(err) { // since we could not allocate, we should put it back // but not if it's a conflict, as the cache is no longer up to date, and // we should wait for it to get updated with fresh info. diff --git a/pkg/gameserverallocations/controller.go b/pkg/gameserverallocations/controller.go index 56795206e9..e1550cfd96 100644 --- a/pkg/gameserverallocations/controller.go +++ b/pkg/gameserverallocations/controller.go @@ -23,7 +23,6 @@ import ( gwruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" "github.com/heptiolabs/healthcheck" - "github.com/pkg/errors" "github.com/sirupsen/logrus" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -46,6 +45,7 @@ import ( "agones.dev/agones/pkg/gameserverallocations/processor" "agones.dev/agones/pkg/gameservers" "agones.dev/agones/pkg/util/apiserver" + "agones.dev/agones/pkg/util/errors" "agones.dev/agones/pkg/util/https" "agones.dev/agones/pkg/util/runtime" ) @@ -61,6 +61,7 @@ type Extensions struct { recorder record.EventRecorder allocator *Allocator processorClient processor.Client + errs *errors.Errors } // NewExtensions returns the extensions controller for a GameServerAllocation @@ -90,6 +91,7 @@ func NewExtensions(apiServer *apiserver.APIServer, allocationBatchWaitTime) c.baseLogger = runtime.NewLoggerWithType(c) + c.errs = errors.FromStruct(c) eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(c.baseLogger.Debugf) @@ -107,6 +109,7 @@ func NewProcessorExtensions(apiServer *apiserver.APIServer, kubeClient kubernete } c.baseLogger = runtime.NewLoggerWithType(c) + c.errs = errors.FromStruct(c) eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(c.baseLogger.Debugf) @@ -212,7 +215,7 @@ func (c *Extensions) allocationDeserialization(r *http.Request, namespace string gvks, _, err := scheme.Scheme.ObjectKinds(gsa) if err != nil { - return gsa, errors.Wrap(err, "error getting objectkinds for gameserverallocation") + return gsa, c.errs.Wrap(err, "error getting objectkinds for gameserverallocation") } gsa.TypeMeta = metav1.TypeMeta{Kind: gvks[0].Kind, APIVersion: gvks[0].Version} @@ -220,23 +223,23 @@ func (c *Extensions) allocationDeserialization(r *http.Request, namespace string mediaTypes := scheme.Codecs.SupportedMediaTypes() mt, _, err := mime.ParseMediaType(r.Header.Get("Content-Type")) if err != nil { - return gsa, errors.Wrap(err, "error parsing mediatype from a request header") + return gsa, c.errs.Wrap(err, "error parsing mediatype from a request header") } info, ok := k8sruntime.SerializerInfoForMediaType(mediaTypes, mt) if !ok { - return gsa, errors.New("Could not find deserializer") + return gsa, c.errs.New("Could not find deserializer") } b, err := io.ReadAll(r.Body) if err != nil { - return gsa, errors.Wrap(err, "could not read body") + return gsa, c.errs.Wrap(err, "could not read body") } gvk := allocationv1.SchemeGroupVersion.WithKind("GameServerAllocation") _, _, err = info.Serializer.Decode(b, &gvk, gsa) if err != nil { c.baseLogger.WithField("body", string(b)).Error("error decoding body") - return gsa, errors.Wrap(err, "error decoding body") + return gsa, c.errs.Wrap(err, "error decoding body") } gsa.ObjectMeta.Namespace = namespace @@ -250,7 +253,7 @@ func (c *Extensions) allocationDeserialization(r *http.Request, namespace string func (c *Extensions) serialisation(r *http.Request, w http.ResponseWriter, obj k8sruntime.Object, statusCode int, codecs serializer.CodecFactory) error { info, err := apiserver.AcceptedSerializer(r, codecs) if err != nil { - return errors.Wrapf(err, "failed to find serialisation info for %T object", obj) + return c.errs.Wrapf(err, "failed to find serialisation info for %T object", obj) } w.Header().Set("Content-Type", info.MediaType) @@ -259,7 +262,7 @@ func (c *Extensions) serialisation(r *http.Request, w http.ResponseWriter, obj k w.WriteHeader(statusCode) err = info.Serializer.Encode(obj, w) - return errors.Wrapf(err, "error encoding %T", obj) + return c.errs.Wrapf(err, "error encoding %T", obj) } // convertProcessorError handles processor client errors and converts them to appropriate responses diff --git a/pkg/gameserverallocations/controller_test.go b/pkg/gameserverallocations/controller_test.go index 96bf48b12d..9893242101 100644 --- a/pkg/gameserverallocations/controller_test.go +++ b/pkg/gameserverallocations/controller_test.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "net/http" "net/http/httptest" @@ -34,7 +35,6 @@ import ( "agones.dev/agones/pkg/util/apiserver" "agones.dev/agones/pkg/util/runtime" "github.com/heptiolabs/healthcheck" - "github.com/pkg/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" @@ -552,7 +552,7 @@ func TestMultiClusterAllocationFromRemote(t *testing.T) { _, err = executeAllocation(gsa, c) if assert.Error(t, err) { - assert.Contains(t, err.Error(), "test error message") + assert.ErrorContains(t, err, "test error message") } assert.Greaterf(t, retry, 1, "Retry count %v. Expecting to retry on error.", retry) }) @@ -788,7 +788,10 @@ func executeAllocation(gsa *allocationv1.GameServerAllocation, c *Extensions) (* ret := &allocationv1.GameServerAllocation{} jsn := rec.Body.Bytes() err = json.Unmarshal(jsn, ret) - return ret, errors.Wrapf(err, "failed to unmarshal allocation response: %s", jsn) + if err != nil { + return ret, fmt.Errorf("failed to unmarshal allocation response: %s: %w", jsn, err) + } + return ret, nil } func addReactorForGameServer(m *agtesting.Mocks) string { diff --git a/pkg/gameserverallocations/find.go b/pkg/gameserverallocations/find.go index 6a39847966..3561491e53 100644 --- a/pkg/gameserverallocations/find.go +++ b/pkg/gameserverallocations/find.go @@ -21,7 +21,6 @@ import ( agonesv1 "agones.dev/agones/pkg/apis/agones/v1" allocationv1 "agones.dev/agones/pkg/apis/allocation/v1" "agones.dev/agones/pkg/util/runtime" - "github.com/pkg/errors" ) // findGameServerForAllocation finds an optimal gameserver, given the @@ -77,7 +76,7 @@ func findGameServerForAllocation(gsa *allocationv1.GameServerAllocation, list [] } } default: - return nil, -1, errors.Errorf("scheduling strategy of '%s' is not supported", gsa.Spec.Scheduling) + return nil, -1, errs.Errorf("scheduling strategy of '%s' is not supported", gsa.Spec.Scheduling) } loop(list, func(i int, gs *agonesv1.GameServer) { diff --git a/pkg/gameserverallocations/gameserverallocations.go b/pkg/gameserverallocations/gameserverallocations.go new file mode 100644 index 0000000000..4564ba538a --- /dev/null +++ b/pkg/gameserverallocations/gameserverallocations.go @@ -0,0 +1,19 @@ +// Copyright Contributors to Agones a Series of LF Projects, LLC. +// +// 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 gameserverallocations + +import "agones.dev/agones/pkg/util/errors" + +var errs = errors.FromPackage() diff --git a/pkg/gameserverallocations/processor/client.go b/pkg/gameserverallocations/processor/client.go index 9b8a437952..122dbdab6e 100644 --- a/pkg/gameserverallocations/processor/client.go +++ b/pkg/gameserverallocations/processor/client.go @@ -20,7 +20,7 @@ import ( "sync" "time" - "github.com/pkg/errors" + "agones.dev/agones/pkg/util/errors" "github.com/sirupsen/logrus" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -83,6 +83,7 @@ type client struct { batchMutex sync.RWMutex // requestIDMapping is a map to correlate request IDs to pendingRequest objects for response handling requestIDMapping map[string]*pendingRequest + errs *errors.Errors } // pendingRequest represents a request waiting for processing @@ -119,7 +120,7 @@ func NewClient(config Config, logger logrus.FieldLogger) Client { config.ClientID = string(uuid.NewUUID()) } - return &client{ + c := &client{ config: config, logger: logger, hotBatch: &allocationpb.BatchRequest{ @@ -128,6 +129,8 @@ func NewClient(config Config, logger logrus.FieldLogger) Client { pendingRequests: make([]*pendingRequest, 0, config.MaxBatchSize), requestIDMapping: make(map[string]*pendingRequest), } + c.errs = errors.FromStruct(c) + return c } // Run starts the processor client and manages the connection lifecycle @@ -231,7 +234,7 @@ func (p *client) handleStream(ctx context.Context, stream allocationpb.Processor return ctx.Err() } p.logger.WithError(err).Error("Failed to receive message from processor") - return errors.Wrap(err, "stream recv error") + return p.errs.Wrap(err, "stream recv error") } // Handle message based on its payload type @@ -481,7 +484,7 @@ func (p *client) connectAndRun(ctx context.Context) error { // Connect to the processor conn, err := p.connect(ctx) if err != nil { - return errors.Wrap(err, "failed to connect") + return p.errs.Wrap(err, "failed to connect") } defer func() { _ = conn.Close() }() @@ -491,12 +494,12 @@ func (p *client) connectAndRun(ctx context.Context) error { // Open a streaming RPC to the processor stream, err := client.StreamBatches(ctx) if err != nil { - return errors.Wrap(err, "failed to create stream") + return p.errs.Wrap(err, "failed to create stream") } // Register this client instance with the processor if err := p.registerClient(stream); err != nil { - return errors.Wrap(err, "failed to register") + return p.errs.Wrap(err, "failed to register") } p.logger.Info("Connected to processor") @@ -546,7 +549,7 @@ func (p *client) healthCheck(ctx context.Context, conn *grpc.ClientConn) error { } if resp.Status != grpc_health_v1.HealthCheckResponse_SERVING { - return errors.Errorf("processor not serving: %v", resp.Status) + return p.errs.Errorf("processor not serving: %v", resp.Status) } return nil