feat(grpc): allow files to be read by jmespath expressions

This allows allow jmespath expressions to refer to the contents of
certain pre-declared files which are reloaded on a user-specified
schedule. For example, if you want to use k8s bound service
account tokens you could add the following to your Pod:

  volumes:
  - name: token-vol
    projected:
      sources:
      - serviceAccountToken:
          audience: my-buildbarn-instance
          expirationSeconds: 3600
          path: buildbarn

Assuming this is mounted at '/tokens', you could then specify in your
buildbarn config:

  addMetadataJmespathExpression: {
    expression: |||
      {
        "authorization": [std.format('bearer %s', files.token)]
      }
    |||,
    files: [
      {
        key: "token",
        path: "/tokens/buildbarn",
        refreshInterval: "1800s",
      }
    ]
  },

This is quite useful for k8s service account tokens, as the maximum
validity is often capped. Likewise this can also be used for Google
service account id tokens, which also have a relatively short maximum
validity.

Author: tomgr

cmd/bb_copy/main.go

@@ -42,6 +42,7 @@ func main() {
 		grpcClientFactory := grpc.NewBaseClientFactory(grpc.BaseClientDialer, nil, nil, nil)
 
 		blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
+			dependenciesGroup,
 			grpcClientFactory,
 			int(configuration.MaximumMessageSizeBytes))
 		source, err := blobstore_configuration.NewBlobAccessFromConfiguration(
@@ -62,7 +63,7 @@ func main() {
 			configuration.Replicator,
 			source.BlobAccess,
 			sink,
-			blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
+			blobstore_configuration.NewCASBlobReplicatorCreator(dependenciesGroup, grpcClientFactory),
 		)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to create replicator")

cmd/bb_replicator/main.go

@@ -27,12 +27,13 @@ func main() {
 		if err := util.UnmarshalConfigurationFromFile(os.Args[1], &configuration); err != nil {
 			return util.StatusWrapf(err, "Failed to read configuration from %s", os.Args[1])
 		}
-		lifecycleState, grpcClientFactory, err := global.ApplyConfiguration(configuration.Global)
+		lifecycleState, grpcClientFactory, err := global.ApplyConfiguration(configuration.Global, dependenciesGroup)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to apply global configuration options")
 		}
 
 		blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
+			dependenciesGroup,
 			grpcClientFactory,
 			int(configuration.MaximumMessageSizeBytes))
 		source, err := blobstore_configuration.NewBlobAccessFromConfiguration(
@@ -53,7 +54,7 @@ func main() {
 			configuration.Replicator,
 			source.BlobAccess,
 			sink,
-			blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
+			blobstore_configuration.NewCASBlobReplicatorCreator(dependenciesGroup, grpcClientFactory),
 		)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to create replicator")

cmd/bb_storage/main.go

@@ -36,7 +36,7 @@ func main() {
 		if err := util.UnmarshalConfigurationFromFile(os.Args[1], &configuration); err != nil {
 			return util.StatusWrapf(err, "Failed to read configuration from %s", os.Args[1])
 		}
-		lifecycleState, grpcClientFactory, err := global.ApplyConfiguration(configuration.Global)
+		lifecycleState, grpcClientFactory, err := global.ApplyConfiguration(configuration.Global, dependenciesGroup)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to apply global configuration options")
 		}
@@ -56,6 +56,7 @@ func main() {
 				dependenciesGroup,
 				configuration.ContentAddressableStorage,
 				blobstore_configuration.NewCASBlobAccessCreator(
+					dependenciesGroup,
 					grpcClientFactory,
 					int(configuration.MaximumMessageSizeBytes)),
 				grpcClientFactory)
@@ -83,9 +84,11 @@ func main() {
 				dependenciesGroup,
 				configuration.ActionCache,
 				blobstore_configuration.NewACBlobAccessCreator(
+					dependenciesGroup,
 					contentAddressableStorageInfo,
 					grpcClientFactory,
-					int(configuration.MaximumMessageSizeBytes)),
+					int(configuration.MaximumMessageSizeBytes),
+				),
 				grpcClientFactory)
 			if err != nil {
 				return util.StatusWrap(err, "Failed to create Action Cache")
@@ -104,6 +107,7 @@ func main() {
 				dependenciesGroup,
 				configuration.IndirectContentAddressableStorage,
 				blobstore_configuration.NewICASBlobAccessCreator(
+					dependenciesGroup,
 					grpcClientFactory,
 					int(configuration.MaximumMessageSizeBytes)),
 				grpcClientFactory)
@@ -120,6 +124,7 @@ func main() {
 				dependenciesGroup,
 				configuration.InitialSizeClassCache,
 				blobstore_configuration.NewISCCBlobAccessCreator(
+					dependenciesGroup,
 					grpcClientFactory,
 					int(configuration.MaximumMessageSizeBytes)),
 				grpcClientFactory)
@@ -136,6 +141,7 @@ func main() {
 				dependenciesGroup,
 				configuration.FileSystemAccessCache,
 				blobstore_configuration.NewFSACBlobAccessCreator(
+					dependenciesGroup,
 					grpcClientFactory,
 					int(configuration.MaximumMessageSizeBytes)),
 				grpcClientFactory)
@@ -158,11 +164,11 @@ func main() {
 		// one or more schedulers specified in the configuration file.
 		var buildQueue builder.BuildQueue
 		if len(configuration.Schedulers) > 0 {
-			baseBuildQueue, err := builder.NewDemultiplexingBuildQueueFromConfiguration(configuration.Schedulers, grpcClientFactory)
+			baseBuildQueue, err := builder.NewDemultiplexingBuildQueueFromConfiguration(configuration.Schedulers, dependenciesGroup, grpcClientFactory)
 			if err != nil {
 				return err
 			}
-			executeAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetExecuteAuthorizer(), grpcClientFactory)
+			executeAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetExecuteAuthorizer(), dependenciesGroup, grpcClientFactory)
 			if err != nil {
 				return util.StatusWrap(err, "Failed to create execute authorizer")
 			}
@@ -240,11 +246,11 @@ func newNonScannableBlobAccess(dependenciesGroup program.Group, configuration *b
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, nil, err
 	}
 
-	getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetAuthorizer, grpcClientFactory)
+	getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetAuthorizer, dependenciesGroup, grpcClientFactory)
 	if err != nil {
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, nil, util.StatusWrap(err, "Failed to create Get() authorizer")
 	}
-	putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.PutAuthorizer, grpcClientFactory)
+	putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.PutAuthorizer, dependenciesGroup, grpcClientFactory)
 	if err != nil {
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, nil, util.StatusWrap(err, "Failed to create Put() authorizer")
 	}
@@ -262,15 +268,15 @@ func newScannableBlobAccess(dependenciesGroup program.Group, configuration *bb_s
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, err
 	}
 
-	getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetAuthorizer, grpcClientFactory)
+	getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.GetAuthorizer, dependenciesGroup, grpcClientFactory)
 	if err != nil {
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, util.StatusWrap(err, "Failed to create Get() authorizer")
 	}
-	putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.PutAuthorizer, grpcClientFactory)
+	putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.PutAuthorizer, dependenciesGroup, grpcClientFactory)
 	if err != nil {
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, util.StatusWrap(err, "Failed to create Put() authorizer")
 	}
-	findMissingAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.FindMissingAuthorizer, grpcClientFactory)
+	findMissingAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.FindMissingAuthorizer, dependenciesGroup, grpcClientFactory)
 	if err != nil {
 		return blobstore_configuration.BlobAccessInfo{}, nil, nil, util.StatusWrap(err, "Failed to create FindMissing() authorizer")
 	}

internal/mock/BUILD.bazel

@@ -410,6 +410,7 @@ go_library(
         "//pkg/digest",
         "//pkg/filesystem",
         "//pkg/filesystem/path",
+        "//pkg/program",
         "//pkg/proto/blobstore/local",
         "//pkg/proto/configuration/grpc",
         "//pkg/util",

pkg/auth/BUILD.bazel

@@ -18,10 +18,10 @@ go_library(
         "//pkg/clock",
         "//pkg/digest",
         "//pkg/eviction",
+        "//pkg/jmespath",
         "//pkg/otel",
         "//pkg/proto/auth",
         "//pkg/util",
-        "@com_github_jmespath_go_jmespath//:go-jmespath",
         "@io_opentelemetry_go_otel//attribute",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//codes",
@@ -47,10 +47,10 @@ go_test(
         "//internal/mock",
         "//pkg/digest",
         "//pkg/eviction",
+        "//pkg/jmespath",
         "//pkg/proto/auth",
         "//pkg/testutil",
         "//pkg/util",
-        "@com_github_jmespath_go_jmespath//:go-jmespath",
         "@com_github_stretchr_testify//require",
         "@io_opentelemetry_go_otel//attribute",
         "@io_opentelemetry_go_proto_otlp//common/v1:common",

pkg/auth/configuration/BUILD.bazel

@@ -11,9 +11,10 @@ go_library(
         "//pkg/digest",
         "//pkg/eviction",
         "//pkg/grpc",
+        "//pkg/jmespath",
+        "//pkg/program",
         "//pkg/proto/configuration/auth",
         "//pkg/util",
-        "@com_github_jmespath_go_jmespath//:go-jmespath",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//status",
         "@org_golang_google_protobuf//encoding/protojson",

pkg/auth/configuration/authorizer_factory.go

@@ -6,9 +6,10 @@ import (
 	"github.com/buildbarn/bb-storage/pkg/digest"
 	"github.com/buildbarn/bb-storage/pkg/eviction"
 	"github.com/buildbarn/bb-storage/pkg/grpc"
+	"github.com/buildbarn/bb-storage/pkg/jmespath"
+	"github.com/buildbarn/bb-storage/pkg/program"
 	pb "github.com/buildbarn/bb-storage/pkg/proto/configuration/auth"
 	"github.com/buildbarn/bb-storage/pkg/util"
-	"github.com/jmespath/go-jmespath"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -20,7 +21,7 @@ import (
 type AuthorizerFactory interface {
 	// NewAuthorizerFromConfiguration constructs an authorizer based on
 	// options specified in a configuration message.
-	NewAuthorizerFromConfiguration(configuration *pb.AuthorizerConfiguration, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error)
+	NewAuthorizerFromConfiguration(configuration *pb.AuthorizerConfiguration, group program.Group, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error)
 }
 
 // DefaultAuthorizerFactory constructs deduplicated authorizers based on
@@ -33,7 +34,7 @@ type BaseAuthorizerFactory struct{}
 
 // NewAuthorizerFromConfiguration constructs an authorizer based on
 // options specified in a configuration message.
-func (f BaseAuthorizerFactory) NewAuthorizerFromConfiguration(config *pb.AuthorizerConfiguration, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error) {
+func (f BaseAuthorizerFactory) NewAuthorizerFromConfiguration(config *pb.AuthorizerConfiguration, group program.Group, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error) {
 	if config == nil {
 		return nil, status.Error(codes.InvalidArgument, "Authorizer configuration not specified")
 	}
@@ -53,13 +54,13 @@ func (f BaseAuthorizerFactory) NewAuthorizerFromConfiguration(config *pb.Authori
 		}
 		return auth.NewStaticAuthorizer(trie.ContainsPrefix), nil
 	case *pb.AuthorizerConfiguration_JmespathExpression:
-		expression, err := jmespath.Compile(policy.JmespathExpression)
+		expression, err := jmespath.NewExpressionFromConfiguration(policy.JmespathExpression, group, clock.SystemClock)
 		if err != nil {
 			return nil, util.StatusWrapWithCode(err, codes.InvalidArgument, "Failed to compile JMESPath expression")
 		}
 		return auth.NewJMESPathExpressionAuthorizer(expression), nil
 	case *pb.AuthorizerConfiguration_Remote:
-		grpcClient, err := grpcClientFactory.NewClientFromConfiguration(policy.Remote.Endpoint)
+		grpcClient, err := grpcClientFactory.NewClientFromConfiguration(policy.Remote.Endpoint, group)
 		if err != nil {
 			return nil, util.StatusWrap(err, "Failed to create authorizer RPC client")
 		}
@@ -96,14 +97,14 @@ func NewDeduplicatingAuthorizerFactory(base AuthorizerFactory) AuthorizerFactory
 }
 
 // NewAuthorizerFromConfiguration creates an Authorizer based on the passed configuration.
-func (af *deduplicatingAuthorizerFactory) NewAuthorizerFromConfiguration(config *pb.AuthorizerConfiguration, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error) {
+func (af *deduplicatingAuthorizerFactory) NewAuthorizerFromConfiguration(config *pb.AuthorizerConfiguration, group program.Group, grpcClientFactory grpc.ClientFactory) (auth.Authorizer, error) {
 	keyBytes, err := protojson.Marshal(config)
 	key := string(keyBytes)
 	if err != nil {
 		return nil, err
 	}
 	if _, ok := af.known[key]; !ok {
-		a, err := af.base.NewAuthorizerFromConfiguration(config, grpcClientFactory)
+		a, err := af.base.NewAuthorizerFromConfiguration(config, group, grpcClientFactory)
 		if err != nil {
 			return nil, err
 		}

pkg/auth/jmespath_expression_authorizer.go

@@ -4,18 +4,18 @@ import (
 	"context"
 
 	"github.com/buildbarn/bb-storage/pkg/digest"
-	"github.com/jmespath/go-jmespath"
+	"github.com/buildbarn/bb-storage/pkg/jmespath"
 )
 
 type jmespathExpressionAuthorizer struct {
-	expression *jmespath.JMESPath
+	expression *jmespath.Expression
 }
 
 // NewJMESPathExpressionAuthorizer creates an Authorizer that evaluates
 // a JMESPath expression to make an authorization decision. The JMESpath
 // expression is called with a JSON object that includes both the REv2
 // instance name and authentication metadata.
-func NewJMESPathExpressionAuthorizer(expression *jmespath.JMESPath) Authorizer {
+func NewJMESPathExpressionAuthorizer(expression *jmespath.Expression) Authorizer {
 	return &jmespathExpressionAuthorizer{
 		expression: expression,
 	}

pkg/auth/jmespath_expression_authorizer_test.go

@@ -6,10 +6,10 @@ import (
 
 	"github.com/buildbarn/bb-storage/pkg/auth"
 	"github.com/buildbarn/bb-storage/pkg/digest"
+	"github.com/buildbarn/bb-storage/pkg/jmespath"
 	auth_pb "github.com/buildbarn/bb-storage/pkg/proto/auth"
 	"github.com/buildbarn/bb-storage/pkg/testutil"
 	"github.com/buildbarn/bb-storage/pkg/util"
-	"github.com/jmespath/go-jmespath"
 	"github.com/stretchr/testify/require"
 
 	"google.golang.org/grpc/codes"

pkg/blobstore/configuration/ac_blob_access_creator.go

@@ -9,6 +9,7 @@ import (
 	"github.com/buildbarn/bb-storage/pkg/clock"
 	"github.com/buildbarn/bb-storage/pkg/digest"
 	"github.com/buildbarn/bb-storage/pkg/grpc"
+	"github.com/buildbarn/bb-storage/pkg/program"
 	pb "github.com/buildbarn/bb-storage/pkg/proto/configuration/blobstore"
 	"github.com/buildbarn/bb-storage/pkg/util"
 
@@ -33,16 +34,18 @@ type acBlobAccessCreator struct {
 	contentAddressableStorage *BlobAccessInfo
 	grpcClientFactory         grpc.ClientFactory
 	maximumMessageSizeBytes   int
+	terminationGroup          program.Group
 }
 
 // NewACBlobAccessCreator creates a BlobAccessCreator that can be
 // provided to NewBlobAccessFromConfiguration() to construct a
 // BlobAccess that is suitable for accessing the Action Cache.
-func NewACBlobAccessCreator(contentAddressableStorage *BlobAccessInfo, grpcClientFactory grpc.ClientFactory, maximumMessageSizeBytes int) BlobAccessCreator {
+func NewACBlobAccessCreator(terminationGroup program.Group, contentAddressableStorage *BlobAccessInfo, grpcClientFactory grpc.ClientFactory, maximumMessageSizeBytes int) BlobAccessCreator {
 	return &acBlobAccessCreator{
 		contentAddressableStorage: contentAddressableStorage,
 		grpcClientFactory:         grpcClientFactory,
 		maximumMessageSizeBytes:   maximumMessageSizeBytes,
+		terminationGroup:          terminationGroup,
 	}
 }
 
@@ -105,7 +108,7 @@ func (bac *acBlobAccessCreator) NewCustomBlobAccess(configuration *pb.BlobAccess
 			DigestKeyFormat: base.DigestKeyFormat.Combine(bac.contentAddressableStorage.DigestKeyFormat),
 		}, "completeness_checking", nil
 	case *pb.BlobAccessConfiguration_Grpc:
-		client, err := bac.grpcClientFactory.NewClientFromConfiguration(backend.Grpc)
+		client, err := bac.grpcClientFactory.NewClientFromConfiguration(backend.Grpc, bac.terminationGroup)
 		if err != nil {
 			return BlobAccessInfo{}, "", err
 		}