ce: feat(server): Add FilteredWithFeatures fn and new server options

Author: hugoghx

internal/server/options.go

@@ -23,37 +23,46 @@ func GetOpts(opt ...Option) options {
 // Option - how Options are passed as arguments
 type Option func(*options)
 
+// StorageBucketCredentialInfo defines the parameters to pass into the
+// WithFilterWorkersByStorageBucketCredentialState option.
+type StorageBucketCredentialInfo struct {
+	CredentialId string
+	Filters      []FilterStorageBucketCredentialStateFn
+}
+
 // options = how options are represented
 type options struct {
-	withName                               string
-	withPublicId                           string
-	withDescription                        string
-	withAddress                            string
-	withLimit                              int
-	withLiveness                           time.Duration
-	withUpdateTags                         bool
-	withWorkerTags                         []*Tag
-	withWorkerKeyIdentifier                string
-	withWorkerKeys                         WorkerKeys
-	withControllerEncryptionPrivateKey     []byte
-	withKeyId                              string
-	withNonce                              []byte
-	withNewIdFunc                          func(context.Context) (string, error)
-	WithFetchNodeCredentialsRequest        *types.FetchNodeCredentialsRequest
-	withTestPkiWorkerAuthorized            bool
-	withTestPkiWorkerKeyId                 *string
-	withTestUseInputTagsAsApiTags          bool
-	withWorkerType                         WorkerType
-	withRoot                               RootInfo
-	withStopAfter                          uint
-	WithCreateControllerLedActivationToken bool
-	withReleaseVersion                     string
-	withOperationalState                   string
-	withLocalStorageState                  string
-	withActiveWorkers                      bool
-	withFeature                            version.Feature
-	withDirectlyConnected                  bool
-	withWorkerPool                         []string
+	withName                                        string
+	withPublicId                                    string
+	withDescription                                 string
+	withAddress                                     string
+	withLimit                                       int
+	withLiveness                                    time.Duration
+	withUpdateTags                                  bool
+	withWorkerTags                                  []*Tag
+	withWorkerKeyIdentifier                         string
+	withWorkerKeys                                  WorkerKeys
+	withControllerEncryptionPrivateKey              []byte
+	withKeyId                                       string
+	withNonce                                       []byte
+	withNewIdFunc                                   func(context.Context) (string, error)
+	WithFetchNodeCredentialsRequest                 *types.FetchNodeCredentialsRequest
+	withTestPkiWorkerAuthorized                     bool
+	withTestPkiWorkerKeyId                          *string
+	withTestUseInputTagsAsApiTags                   bool
+	withWorkerType                                  WorkerType
+	withRoot                                        RootInfo
+	withStopAfter                                   uint
+	WithCreateControllerLedActivationToken          bool
+	withReleaseVersion                              string
+	withOperationalState                            string
+	withLocalStorageState                           string
+	withActiveWorkers                               bool
+	withFeature                                     version.Feature
+	withDirectlyConnected                           bool
+	withWorkerPool                                  []string
+	withFilterWorkersByStorageBucketCredentialState *StorageBucketCredentialInfo
+	withFilterWorkersByLocalStorageState            bool
 }
 
 func getDefaultOptions() options {
@@ -276,3 +285,21 @@ func WithLocalStorageState(state string) Option {
 		o.withLocalStorageState = state
 	}
 }
+
+// WithFilterWorkersByStorageBucketCredentialState receives a storage bucket
+// credential id and filters to apply and calls
+// FilterWorkersByStorageBucketCredentialState in supported repository
+// functions.
+func WithFilterWorkersByStorageBucketCredentialState(ci *StorageBucketCredentialInfo) Option {
+	return func(o *options) {
+		o.withFilterWorkersByStorageBucketCredentialState = ci
+	}
+}
+
+// WithFilterWorkersByLocalStorageState controls whether
+// FilterWorkersByLocalStorageState is called in supported repository functions.
+func WithFilterWorkersByLocalStorageState(filter bool) Option {
+	return func(o *options) {
+		o.withFilterWorkersByLocalStorageState = filter
+	}
+}

internal/server/worker_list.go

@@ -4,14 +4,18 @@
 package server
 
 import (
+	"context"
 	stderrors "errors"
 	"fmt"
+	"strings"
 
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
+	"github.com/hashicorp/boundary/internal/event"
 	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
 	"github.com/hashicorp/boundary/sdk/pbs/plugin"
 	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-bexpr"
+	gvers "github.com/hashicorp/go-version"
 	"github.com/mitchellh/pointerstructure"
 	"google.golang.org/grpc/codes"
 )
@@ -84,6 +88,51 @@ func (w WorkerList) Filtered(eval *bexpr.Evaluator) (WorkerList, error) {
 	return ret, nil
 }
 
+// FilteredWithFeatures returns a new workerList where all elements contained in
+// it are the ones which from the original workerList that pass the evaluator's
+// evaluation and satisfy the features required.
+func (w WorkerList) FilteredWithFeatures(ctx context.Context, eval *bexpr.Evaluator, features []version.Feature) (WorkerList, error) {
+	const op = "server.WorkerList.FilteredWithFeatures"
+	var ret []*Worker
+workerLoop:
+	for _, worker := range w {
+		filterInput := map[string]interface{}{
+			"name": worker.GetName(),
+			"tags": worker.CanonicalTags(),
+		}
+		ok, err := eval.Evaluate(filterInput)
+		if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
+			// If we find pointerstructure.ErrNotFound, don't error out but
+			// ignore the worker and go to the next.
+			return nil, err
+		}
+		if !ok {
+			continue
+		}
+		if len(features) > 0 {
+			versionString := worker.ReleaseVersion
+			idx := strings.Index(versionString, version.BoundaryPrefix)
+			if idx >= 0 {
+				versionString = versionString[idx+len(version.BoundaryPrefix):]
+			}
+			nodeVersion, err := gvers.NewVersion(versionString)
+			if err != nil {
+				// Emit error and continue, as we might still find a different worker
+				event.WriteError(ctx, op, fmt.Errorf("cannot parse worker version %s for worker %s", versionString, worker.Name))
+				continue
+			}
+			for _, f := range features {
+				if !version.SupportsFeature(nodeVersion, f) {
+					continue workerLoop
+				}
+			}
+		}
+
+		ret = append(ret, worker)
+	}
+	return ret, nil
+}
+
 // SeparateManagedWorkers divides the incoming workers into managed and
 // unmanaged workers, respectively
 func SeparateManagedWorkers(workers WorkerList) (managedWorkers, nonManagedWorkers WorkerList) {