Merge remote-tracking branch 'upstream/master'

authz/audit_logger.go

@@ -0,0 +1,126 @@
+/*
+ *
+ * Copyright 2023 gRPC authors.
+ *
+ * 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 authz
+
+import (
+	"encoding/json"
+	"sync"
+)
+
+// loggerBuilderRegistry holds a map of audit logger builders and a mutex
+// to facilitate thread-safe reading/writing operations.
+type loggerBuilderRegistry struct {
+	mu       sync.Mutex
+	builders map[string]AuditLoggerBuilder
+}
+
+var (
+	registry = loggerBuilderRegistry{
+		builders: make(map[string]AuditLoggerBuilder),
+	}
+)
+
+// RegisterAuditLoggerBuilder registers the builder in a global map
+// using b.Name() as the key.
+//
+// This should only be called during initialization time (i.e. in an init()
+// function). If multiple builders are registered with the same name,
+// the one registered last will take effect.
+func RegisterAuditLoggerBuilder(b AuditLoggerBuilder) {
+	registry.mu.Lock()
+	defer registry.mu.Unlock()
+	registry.builders[b.Name()] = b
+}
+
+// GetAuditLoggerBuilder returns a builder with the given name.
+// It returns nil if the builder is not found in the registry.
+func GetAuditLoggerBuilder(name string) AuditLoggerBuilder {
+	registry.mu.Lock()
+	defer registry.mu.Unlock()
+	return registry.builders[name]
+}
+
+// AuditEvent contains information passed to the audit logger as part of an
+// audit logging event.
+type AuditEvent struct {
+	// FullMethodName is the full method name of the audited RPC, in the format
+	// of "/pkg.Service/Method". For example, "/helloworld.Greeter/SayHello".
+	FullMethodName string
+	// Principal is the identity of the caller. Currently it will only be
+	// available in certificate-based TLS authentication.
+	Principal string
+	// PolicyName is the authorization policy name or the xDS RBAC filter name.
+	PolicyName string
+	// MatchedRule is the matched rule or policy name in the xDS RBAC filter.
+	// It will be empty if there is no match.
+	MatchedRule string
+	// Authorized indicates whether the audited RPC is authorized or not.
+	Authorized bool
+}
+
+// AuditLoggerConfig represents an opaque data structure holding an audit
+// logger configuration. Concrete types representing configuration of specific
+// audit loggers must embed this interface to implement it.
+type AuditLoggerConfig interface {
+	auditLoggerConfig()
+}
+
+// AuditLogger is the interface to be implemented by audit loggers.
+//
+// An audit logger is a logger instance that can be configured via the
+// authorization policy API or xDS HTTP RBAC filters. When the authorization
+// decision meets the condition for audit, all the configured audit loggers'
+// Log() method will be invoked to log that event.
+//
+// TODO(lwge): Change the link to the merged gRFC once it's ready.
+// Please refer to https://github.com/grpc/proposal/pull/346 for more details
+// about audit logging.
+type AuditLogger interface {
+	// Log performs audit logging for the provided audit event.
+	//
+	// This method is invoked in the RPC path and therefore implementations
+	// must not block.
+	Log(*AuditEvent)
+}
+
+// AuditLoggerBuilder is the interface to be implemented by audit logger
+// builders that are used at runtime to configure and instantiate audit loggers.
+//
+// Users who want to implement their own audit logging logic should
+// implement this interface, along with the AuditLogger interface, and register
+// it by calling RegisterAuditLoggerBuilder() at init time.
+//
+// TODO(lwge): Change the link to the merged gRFC once it's ready.
+// Please refer to https://github.com/grpc/proposal/pull/346 for more details
+// about audit logging.
+type AuditLoggerBuilder interface {
+	// ParseAuditLoggerConfig parses the given JSON bytes into a structured
+	// logger config this builder can use to build an audit logger.
+	ParseAuditLoggerConfig(config json.RawMessage) (AuditLoggerConfig, error)
+	// Build builds an audit logger with the given logger config.
+	// This will only be called with valid configs returned from
+	// ParseAuditLoggerConfig() and any runtime issues such as failing to
+	// create a file should be handled by the logger implementation instead of
+	// failing the logger instantiation. So implementers need to make sure it
+	// can return a logger without error at this stage.
+	Build(AuditLoggerConfig) AuditLogger
+	// Name returns the name of logger built by this builder.
+	// This is used to register and pick the builder.
+	Name() string
+}

balancer/rls/balancer.go

@@ -188,7 +188,10 @@ func (b *rlsBalancer) run() {
 
 	for {
 		select {
-		case u := <-b.updateCh.Get():
+		case u, ok := <-b.updateCh.Get():
+			if !ok {
+				return
+			}
 			b.updateCh.Load()
 			switch update := u.(type) {
 			case childPolicyIDAndState:
@@ -450,6 +453,8 @@ func (b *rlsBalancer) Close() {
 	b.dataCache.stop()
 	b.cacheMu.Unlock()
 
+	b.updateCh.Close()
+
 	<-b.done.Done()
 }
 

balancer_conn_wrappers.go

@@ -113,7 +113,10 @@ type subConnUpdate struct {
 func (ccb *ccBalancerWrapper) watcher() {
 	for {
 		select {
-		case u := <-ccb.updateCh.Get():
+		case u, ok := <-ccb.updateCh.Get():
+			if !ok {
+				break
+			}
 			ccb.updateCh.Load()
 			if ccb.closed.HasFired() {
 				break
@@ -155,8 +158,13 @@ func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnStat
 	ccb.updateCh.Put(&ccStateUpdate{ccs: ccs})
 
 	var res interface{}
+	var ok bool
 	select {
-	case res = <-ccb.resultCh.Get():
+	case res, ok = <-ccb.resultCh.Get():
+		if !ok {
+			// The result channel is closed only when the balancer wrapper is closed.
+			return nil
+		}
 		ccb.resultCh.Load()
 	case <-ccb.closed.Done():
 		// Return early if the balancer wrapper is closed while we are waiting for
@@ -296,6 +304,8 @@ func (ccb *ccBalancerWrapper) close() {
 
 func (ccb *ccBalancerWrapper) handleClose() {
 	ccb.balancer.Close()
+	ccb.updateCh.Close()
+	ccb.resultCh.Close()
 	ccb.done.Fire()
 }
 

benchmark/primitives/code_string_test.go

@@ -87,49 +87,39 @@ func (i codeBench) StringUsingMap() string {
 }
 
 func BenchmarkCodeStringStringer(b *testing.B) {
-	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		c := codeBench(uint32(i % 17))
 		_ = c.String()
 	}
-	b.StopTimer()
 }
 
 func BenchmarkCodeStringMap(b *testing.B) {
-	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		c := codeBench(uint32(i % 17))
 		_ = c.StringUsingMap()
 	}
-	b.StopTimer()
 }
 
 // codes.Code.String() does a switch.
 func BenchmarkCodeStringSwitch(b *testing.B) {
-	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		c := codes.Code(uint32(i % 17))
 		_ = c.String()
 	}
-	b.StopTimer()
 }
 
 // Testing all codes (0<=c<=16) and also one overflow (17).
 func BenchmarkCodeStringStringerWithOverflow(b *testing.B) {
-	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		c := codeBench(uint32(i % 18))
 		_ = c.String()
 	}
-	b.StopTimer()
 }
 
 // Testing all codes (0<=c<=16) and also one overflow (17).
 func BenchmarkCodeStringSwitchWithOverflow(b *testing.B) {
-	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		c := codes.Code(uint32(i % 18))
 		_ = c.String()
 	}
-	b.StopTimer()
 }

clientconn.go

@@ -244,19 +244,6 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 		}
 	}()
 
-	scSet := false
-	if cc.dopts.scChan != nil {
-		// Try to get an initial service config.
-		select {
-		case sc, ok := <-cc.dopts.scChan:
-			if ok {
-				cc.sc = &sc
-				cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{&sc})
-				scSet = true
-			}
-		default:
-		}
-	}
 	if cc.dopts.bs == nil {
 		cc.dopts.bs = backoff.DefaultExponential
 	}
@@ -272,7 +259,7 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 	}
 	channelz.Infof(logger, cc.channelzID, "Channel authority set to %q", cc.authority)
 
-	if cc.dopts.scChan != nil && !scSet {
+	if cc.dopts.scChan != nil {
 		// Blocking wait for the initial service config.
 		select {
 		case sc, ok := <-cc.dopts.scChan: