xds extended event (#2158)

internal/contextkeys/contextkeys.go

@@ -0,0 +1,15 @@
+// Package contextkeys defines common context keys shared between packages
+package contextkeys
+
+type contextKey int
+
+const (
+	// DatabrokerConfigVersion identifies the uint64 databroker version of the config
+	DatabrokerConfigVersion contextKey = iota
+)
+
+func (x contextKey) String() string {
+	return map[contextKey]string{
+		DatabrokerConfigVersion: "db_config_version",
+	}[x]
+}

internal/controlplane/events.go

@@ -10,13 +10,13 @@ import (
 
 	"github.com/pomerium/pomerium/internal/log"
 	"github.com/pomerium/pomerium/pkg/grpc"
-	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
 	databrokerpb "github.com/pomerium/pomerium/pkg/grpc/databroker"
+	"github.com/pomerium/pomerium/pkg/grpc/events"
 )
 
 const maxEnvoyConfigurationEvents = 50
 
-func (srv *Server) handleEnvoyConfigurationEvent(evt *configpb.EnvoyConfigurationEvent) {
+func (srv *Server) handleEnvoyConfigurationEvent(evt *events.EnvoyConfigurationEvent) {
 	select {
 	case srv.envoyConfigurationEvents <- evt:
 	default:
@@ -28,7 +28,7 @@ func (srv *Server) handleEnvoyConfigurationEvent(evt *configpb.EnvoyConfiguratio
 
 func (srv *Server) runEnvoyConfigurationEventHandler(ctx context.Context) error {
 	for {
-		var evt *configpb.EnvoyConfigurationEvent
+		var evt *events.EnvoyConfigurationEvent
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -41,7 +41,7 @@ func (srv *Server) runEnvoyConfigurationEventHandler(ctx context.Context) error
 	}
 }
 
-func (srv *Server) storeEnvoyConfigurationEvent(ctx context.Context, evt *configpb.EnvoyConfigurationEvent) error {
+func (srv *Server) storeEnvoyConfigurationEvent(ctx context.Context, evt *events.EnvoyConfigurationEvent) error {
 	any, err := anypb.New(evt)
 	if err != nil {
 		return err

internal/controlplane/events_test.go

@@ -13,8 +13,8 @@ import (
 
 	"github.com/pomerium/pomerium/config"
 	"github.com/pomerium/pomerium/pkg/cryptutil"
-	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
 	databrokerpb "github.com/pomerium/pomerium/pkg/grpc/databroker"
+	"github.com/pomerium/pomerium/pkg/grpc/events"
 )
 
 type mockDataBrokerServer struct {
@@ -33,8 +33,8 @@ func (mock *mockDataBrokerServer) SetOptions(ctx context.Context, req *databroke
 
 func TestEvents(t *testing.T) {
 	t.Run("passes events", func(t *testing.T) {
-		srv := &Server{envoyConfigurationEvents: make(chan *configpb.EnvoyConfigurationEvent, 1)}
+		srv := &Server{envoyConfigurationEvents: make(chan *events.EnvoyConfigurationEvent, 1)}
-		srv.handleEnvoyConfigurationEvent(new(configpb.EnvoyConfigurationEvent))
+		srv.handleEnvoyConfigurationEvent(new(events.EnvoyConfigurationEvent))
 		evt := <-srv.envoyConfigurationEvents
 		assert.NotNil(t, evt)
 	})
@@ -42,7 +42,7 @@ func TestEvents(t *testing.T) {
 		ctx := context.Background()
 
 		srv := &Server{
-			envoyConfigurationEvents: make(chan *configpb.EnvoyConfigurationEvent, 1),
+			envoyConfigurationEvents: make(chan *events.EnvoyConfigurationEvent, 1),
 		}
 		srv.currentConfig.Store(versionedConfig{
 			Config: &config.Config{
@@ -55,7 +55,7 @@ func TestEvents(t *testing.T) {
 		eg.Go(func() error {
 			return srv.runEnvoyConfigurationEventHandler(ctx)
 		})
-		srv.envoyConfigurationEvents <- new(configpb.EnvoyConfigurationEvent)
+		srv.envoyConfigurationEvents <- new(events.EnvoyConfigurationEvent)
 		cancel()
 		assert.Equal(t, context.Canceled, eg.Wait())
 	})
@@ -107,13 +107,13 @@ func TestEvents(t *testing.T) {
 					},
 				},
 			})
-			err := srv.storeEnvoyConfigurationEvent(ctx, new(configpb.EnvoyConfigurationEvent))
+			err := srv.storeEnvoyConfigurationEvent(ctx, new(events.EnvoyConfigurationEvent))
 			assert.NoError(t, err)
 			return err
 		})
 		_ = eg.Wait()
 
 		assert.Equal(t, uint64(maxEnvoyConfigurationEvents), setOptionsRequest.GetOptions().GetCapacity())
-		assert.Equal(t, "type.googleapis.com/pomerium.config.EnvoyConfigurationEvent", putRequest.GetRecord().GetType())
+		assert.Equal(t, "type.googleapis.com/pomerium.events.EnvoyConfigurationEvent", putRequest.GetRecord().GetType())
 	})
 }

internal/controlplane/server.go

@@ -24,7 +24,7 @@ import (
 	"github.com/pomerium/pomerium/internal/telemetry"
 	"github.com/pomerium/pomerium/internal/telemetry/requestid"
 	"github.com/pomerium/pomerium/internal/version"
-	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
+	"github.com/pomerium/pomerium/pkg/grpc/events"
 	"github.com/pomerium/pomerium/pkg/grpcutil"
 )
 
@@ -61,7 +61,7 @@ type Server struct {
 	reproxy       *reproxy.Handler
 
 	haveSetEnvoyConfigurationEventOptions bool
-	envoyConfigurationEvents              chan *configpb.EnvoyConfigurationEvent
+	envoyConfigurationEvents              chan *events.EnvoyConfigurationEvent
 }
 
 // NewServer creates a new Server. Listener ports are chosen by the OS.
@@ -69,7 +69,7 @@ func NewServer(name string, metricsMgr *config.MetricsManager) (*Server, error)
 	srv := &Server{
 		metricsMgr:               metricsMgr,
 		reproxy:                  reproxy.New(),
-		envoyConfigurationEvents: make(chan *configpb.EnvoyConfigurationEvent, 10),
+		envoyConfigurationEvents: make(chan *events.EnvoyConfigurationEvent, 10),
 	}
 	srv.currentConfig.Store(versionedConfig{
 		Config: &config.Config{Options: &config.Options{}},

internal/controlplane/xdsmgr/xdsmgr.go

@@ -9,14 +9,20 @@ import (
 
 	envoy_service_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	"github.com/google/uuid"
+	lru "github.com/hashicorp/golang-lru"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/pomerium/pomerium/internal/contextkeys"
 	"github.com/pomerium/pomerium/internal/log"
 	"github.com/pomerium/pomerium/internal/signal"
-	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
+	"github.com/pomerium/pomerium/pkg/grpc/events"
+)
+
+const (
+	maxNonceCacheSize = 1 << 12
 )
 
 type streamState struct {
@@ -30,21 +36,26 @@ var onHandleDeltaRequest = func(state *streamState) {}
 // A Manager manages xDS resources.
 type Manager struct {
 	signal       *signal.Signal
-	eventHandler func(*configpb.EnvoyConfigurationEvent)
+	eventHandler func(*events.EnvoyConfigurationEvent)
 
 	mu        sync.Mutex
 	nonce     string
 	resources map[string][]*envoy_service_discovery_v3.Resource
+
+	nonceToConfig *lru.Cache
 }
 
 // NewManager creates a new Manager.
-func NewManager(resources map[string][]*envoy_service_discovery_v3.Resource, eventHandler func(*configpb.EnvoyConfigurationEvent)) *Manager {
+func NewManager(resources map[string][]*envoy_service_discovery_v3.Resource, eventHandler func(*events.EnvoyConfigurationEvent)) *Manager {
+	nonceToConfig, _ := lru.New(maxNonceCacheSize) // the only error they return is when size is negative, which never happens
+
 	return &Manager{
 		signal:       signal.New(),
 		eventHandler: eventHandler,
 
-		nonce:     uuid.New().String(),
+		nonceToConfig: nonceToConfig,
-		resources: resources,
+		nonce:         uuid.NewString(),
+		resources:     resources,
 	}
 }
 
@@ -114,44 +125,25 @@ func (mgr *Manager) DeltaAggregatedResources(
 			// neither an ACK or a NACK
 		case req.GetErrorDetail() != nil:
 			// a NACK
-			bs, _ := json.Marshal(req.ErrorDetail.Details)
-			log.Error(ctx).
-				Err(errors.New(req.ErrorDetail.Message)).
-				Str("nonce", req.ResponseNonce).
-				Int32("code", req.ErrorDetail.Code).
-				RawJSON("details", bs).Msg("error applying configuration")
 			// - set the client resource versions to the current resource versions
 			state.clientResourceVersions = make(map[string]string)
 			for _, resource := range mgr.resources[req.GetTypeUrl()] {
 				state.clientResourceVersions[resource.Name] = resource.Version
 			}
 
-			mgr.eventHandler(&configpb.EnvoyConfigurationEvent{
+			mgr.nackEvent(ctx, req)
-				Time:    timestamppb.Now(),
-				Message: req.ErrorDetail.Message,
-				Code:    req.ErrorDetail.Code,
-				Details: req.ErrorDetail.Details,
-			})
 		case req.GetResponseNonce() == mgr.nonce:
 			// an ACK for the last response
 			// - set the client resource versions to the current resource versions
-			log.Debug(ctx).
-				Str("nonce", req.ResponseNonce).
-				Msg("ACK")
 			state.clientResourceVersions = make(map[string]string)
 			for _, resource := range mgr.resources[req.GetTypeUrl()] {
 				state.clientResourceVersions[resource.Name] = resource.Version
 			}
 
-			mgr.eventHandler(&configpb.EnvoyConfigurationEvent{
+			mgr.ackEvent(ctx, req)
-				Time:    timestamppb.Now(),
-				Message: "OK",
-			})
 		default:
 			// an ACK for a response that's not the last response
-			log.Debug(ctx).
+			mgr.ackEvent(ctx, req)
-				Str("nonce", req.ResponseNonce).
-				Msg("stale ACK")
 		}
 
 		// update subscriptions
@@ -220,10 +212,7 @@ func (mgr *Manager) DeltaAggregatedResources(
 				case <-ctx.Done():
 					return ctx.Err()
 				case outgoing <- res:
-					log.Info(changeCtx).
+					mgr.changeEvent(ctx, res)
-						Str("nonce", res.Nonce).
-						Str("type", res.TypeUrl).
-						Msg("send update")
 				}
 			}
 		}
@@ -255,10 +244,92 @@ func (mgr *Manager) StreamAggregatedResources(
 // Update updates the state of resources. If any changes are made they will be pushed to any listening
 // streams. For each TypeURL the list of resources should be the complete list of resources.
 func (mgr *Manager) Update(ctx context.Context, resources map[string][]*envoy_service_discovery_v3.Resource) {
+	nonce := uuid.New().String()
+
 	mgr.mu.Lock()
-	mgr.nonce = uuid.New().String()
+	mgr.nonce = nonce
 	mgr.resources = resources
+	mgr.nonceToConfig.Add(nonce, ctx.Value(contextkeys.DatabrokerConfigVersion))
 	mgr.mu.Unlock()
 
 	mgr.signal.Broadcast(ctx)
 }
+
+func (mgr *Manager) nonceToConfigVersion(nonce string) (ver uint64) {
+	val, ok := mgr.nonceToConfig.Get(nonce)
+	if !ok {
+		return 0
+	}
+	ver, _ = val.(uint64)
+	return ver
+}
+
+func (mgr *Manager) nackEvent(ctx context.Context, req *envoy_service_discovery_v3.DeltaDiscoveryRequest) {
+	mgr.eventHandler(&events.EnvoyConfigurationEvent{
+		Kind:                 events.EnvoyConfigurationEvent_EVENT_DISCOVERY_REQUEST,
+		Time:                 timestamppb.Now(),
+		Message:              req.ErrorDetail.Message,
+		Code:                 req.ErrorDetail.Code,
+		Details:              req.ErrorDetail.Details,
+		ResourceSubscribed:   req.ResourceNamesSubscribe,
+		ResourceUnsubscribed: req.ResourceNamesUnsubscribe,
+		ConfigVersion:        mgr.nonceToConfigVersion(req.ResponseNonce),
+		TypeUrl:              req.TypeUrl,
+	})
+
+	bs, _ := json.Marshal(req.ErrorDetail.Details)
+	log.Error(ctx).
+		Err(errors.New(req.ErrorDetail.Message)).
+		Str("resource_type", req.TypeUrl).
+		Strs("resources_unsubscribe", req.ResourceNamesUnsubscribe).
+		Strs("resources_subscribe", req.ResourceNamesSubscribe).
+		Uint64("nonce_version", mgr.nonceToConfigVersion(req.ResponseNonce)).
+		Int32("code", req.ErrorDetail.Code).
+		RawJSON("details", bs).Msg("error applying configuration")
+}
+
+func (mgr *Manager) ackEvent(ctx context.Context, req *envoy_service_discovery_v3.DeltaDiscoveryRequest) {
+	mgr.eventHandler(&events.EnvoyConfigurationEvent{
+		Kind:                 events.EnvoyConfigurationEvent_EVENT_DISCOVERY_REQUEST,
+		Time:                 timestamppb.Now(),
+		ConfigVersion:        mgr.nonceToConfigVersion(req.ResponseNonce),
+		ResourceSubscribed:   req.ResourceNamesSubscribe,
+		ResourceUnsubscribed: req.ResourceNamesUnsubscribe,
+		TypeUrl:              req.TypeUrl,
+		Message:              "ok",
+	})
+
+	log.Debug(ctx).
+		Str("resource_type", req.TypeUrl).
+		Strs("resources_unsubscribe", req.ResourceNamesUnsubscribe).
+		Strs("resources_subscribe", req.ResourceNamesSubscribe).
+		Uint64("nonce_version", mgr.nonceToConfigVersion(req.ResponseNonce)).
+		Msg("ACK")
+}
+
+func (mgr *Manager) changeEvent(ctx context.Context, res *envoy_service_discovery_v3.DeltaDiscoveryResponse) {
+	mgr.eventHandler(&events.EnvoyConfigurationEvent{
+		Kind:                 events.EnvoyConfigurationEvent_EVENT_DISCOVERY_RESPONSE,
+		Time:                 timestamppb.Now(),
+		ConfigVersion:        mgr.nonceToConfigVersion(res.Nonce),
+		ResourceSubscribed:   resourceNames(res.Resources),
+		ResourceUnsubscribed: res.RemovedResources,
+		TypeUrl:              res.TypeUrl,
+		Message:              "change",
+	})
+	log.Debug(ctx).
+		Uint64("ctx_config_version", mgr.nonceToConfigVersion(res.Nonce)).
+		Str("nonce", res.Nonce).
+		Str("type", res.TypeUrl).
+		Strs("subscribe", resourceNames(res.Resources)).
+		Strs("removed", res.RemovedResources).
+		Msg("sent update")
+}
+
+func resourceNames(res []*envoy_service_discovery_v3.Resource) []string {
+	txt := make([]string, 0, len(res))
+	for _, r := range res {
+		txt = append(txt, r.Name)
+	}
+	return txt
+}

internal/controlplane/xdsmgr/xdsmgr_test.go

@@ -14,7 +14,7 @@ import (
 	"google.golang.org/grpc/test/bufconn"
 
 	"github.com/pomerium/pomerium/internal/signal"
-	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
+	"github.com/pomerium/pomerium/pkg/grpc/events"
 )
 
 const bufSize = 1024 * 1024
@@ -37,7 +37,7 @@ func TestManager(t *testing.T) {
 		typeURL: {
 			{Name: "r1", Version: "1"},
 		},
-	}, func(evt *configpb.EnvoyConfigurationEvent) {})
+	}, func(evt *events.EnvoyConfigurationEvent) {})
 	envoy_service_discovery_v3.RegisterAggregatedDiscoveryServiceServer(srv, mgr)
 
 	li := bufconn.Listen(bufSize)

pkg/grpc/config/config.proto

@@ -3,10 +3,8 @@ syntax = "proto3";
 package pomerium.config;
 option go_package = "github.com/pomerium/pomerium/pkg/grpc/config";
 
-import "google/protobuf/any.proto";
 import "google/protobuf/duration.proto";
 import "google/protobuf/struct.proto";
-import "google/protobuf/timestamp.proto";
 import "envoy/config/cluster/v3/cluster.proto";
 import "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.proto";
 
@@ -189,16 +187,3 @@ message Settings {
   optional envoy.extensions.filters.network.http_connection_manager.v3
       .HttpConnectionManager.CodecType codec_type = 73;
 }
-
-// EnvoyConfigurationEvents is a list of envoy configuration events.
-message EnvoyConfigurationEvents {
-  repeated EnvoyConfigurationEvent values = 1;
-}
-
-// EnvoyConfigurationEvent is an envoy configuration event.
-message EnvoyConfigurationEvent {
-  google.protobuf.Timestamp time = 1;
-  string message = 2;
-  int32 code = 3;
-  repeated google.protobuf.Any details = 4;
-}

pkg/grpc/databroker/syncer.go

@@ -10,6 +10,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/pomerium/pomerium/internal/contextkeys"
 	"github.com/pomerium/pomerium/internal/log"
 )
 
@@ -176,10 +177,11 @@ func (syncer *Syncer) sync(ctx context.Context) error {
 
 // logCtx adds log params to context which
 func (syncer *Syncer) logCtx(ctx context.Context) context.Context {
-	return log.WithContext(ctx, func(c zerolog.Context) zerolog.Context {
+	ctx = log.WithContext(ctx, func(c zerolog.Context) zerolog.Context {
 		return c.Str("syncer_id", syncer.id).
 			Str("type", syncer.cfg.typeURL).
 			Uint64("server_version", syncer.serverVersion).
 			Uint64("record_version", syncer.recordVersion)
 	})
+	return context.WithValue(ctx, contextkeys.DatabrokerConfigVersion, syncer.recordVersion)
 }

pkg/grpc/events/xds.pb.go

@@ -0,0 +1,384 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.26.0
+// 	protoc        v3.14.0
+// source: xds.proto
+
+package events
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	anypb "google.golang.org/protobuf/types/known/anypb"
+	timestamppb "google.golang.org/protobuf/types/known/timestamppb"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type EnvoyConfigurationEvent_EventKind int32
+
+const (
+	EnvoyConfigurationEvent_EVENT_KIND_UNDEFINED EnvoyConfigurationEvent_EventKind = 0
+	// envoy_service_discovery_v3.DeltaDiscoveryRequest
+	EnvoyConfigurationEvent_EVENT_DISCOVERY_REQUEST EnvoyConfigurationEvent_EventKind = 1
+	// envoy_service_discovery_v3.DeltaDiscoveryResponse
+	EnvoyConfigurationEvent_EVENT_DISCOVERY_RESPONSE EnvoyConfigurationEvent_EventKind = 2
+)
+
+// Enum value maps for EnvoyConfigurationEvent_EventKind.
+var (
+	EnvoyConfigurationEvent_EventKind_name = map[int32]string{
+		0: "EVENT_KIND_UNDEFINED",
+		1: "EVENT_DISCOVERY_REQUEST",
+		2: "EVENT_DISCOVERY_RESPONSE",
+	}
+	EnvoyConfigurationEvent_EventKind_value = map[string]int32{
+		"EVENT_KIND_UNDEFINED":     0,
+		"EVENT_DISCOVERY_REQUEST":  1,
+		"EVENT_DISCOVERY_RESPONSE": 2,
+	}
+)
+
+func (x EnvoyConfigurationEvent_EventKind) Enum() *EnvoyConfigurationEvent_EventKind {
+	p := new(EnvoyConfigurationEvent_EventKind)
+	*p = x
+	return p
+}
+
+func (x EnvoyConfigurationEvent_EventKind) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (EnvoyConfigurationEvent_EventKind) Descriptor() protoreflect.EnumDescriptor {
+	return file_xds_proto_enumTypes[0].Descriptor()
+}
+
+func (EnvoyConfigurationEvent_EventKind) Type() protoreflect.EnumType {
+	return &file_xds_proto_enumTypes[0]
+}
+
+func (x EnvoyConfigurationEvent_EventKind) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use EnvoyConfigurationEvent_EventKind.Descriptor instead.
+func (EnvoyConfigurationEvent_EventKind) EnumDescriptor() ([]byte, []int) {
+	return file_xds_proto_rawDescGZIP(), []int{1, 0}
+}
+
+// EnvoyConfigurationEvents is a list of envoy configuration events.
+type EnvoyConfigurationEvents struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Values []*EnvoyConfigurationEvent `protobuf:"bytes,1,rep,name=values,proto3" json:"values,omitempty"`
+}
+
+func (x *EnvoyConfigurationEvents) Reset() {
+	*x = EnvoyConfigurationEvents{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_xds_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *EnvoyConfigurationEvents) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EnvoyConfigurationEvents) ProtoMessage() {}
+
+func (x *EnvoyConfigurationEvents) ProtoReflect() protoreflect.Message {
+	mi := &file_xds_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use EnvoyConfigurationEvents.ProtoReflect.Descriptor instead.
+func (*EnvoyConfigurationEvents) Descriptor() ([]byte, []int) {
+	return file_xds_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *EnvoyConfigurationEvents) GetValues() []*EnvoyConfigurationEvent {
+	if x != nil {
+		return x.Values
+	}
+	return nil
+}
+
+// EnvoyConfigurationEvent is an envoy configuration event.
+type EnvoyConfigurationEvent struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Time    *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=time,proto3" json:"time,omitempty"`
+	Message string                 `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"`
+	Code    int32                  `protobuf:"varint,3,opt,name=code,proto3" json:"code,omitempty"`
+	Details []*anypb.Any           `protobuf:"bytes,4,rep,name=details,proto3" json:"details,omitempty"`
+	// databroker config version
+	ConfigVersion uint64 `protobuf:"varint,5,opt,name=config_version,json=configVersion,proto3" json:"config_version,omitempty"`
+	// envoy resource type (i.e. listener, cluster)
+	TypeUrl              string                            `protobuf:"bytes,6,opt,name=type_url,json=typeUrl,proto3" json:"type_url,omitempty"`
+	Kind                 EnvoyConfigurationEvent_EventKind `protobuf:"varint,7,opt,name=kind,proto3,enum=pomerium.events.EnvoyConfigurationEvent_EventKind" json:"kind,omitempty"`
+	ResourceSubscribed   []string                          `protobuf:"bytes,8,rep,name=resource_subscribed,json=resourceSubscribed,proto3" json:"resource_subscribed,omitempty"`
+	ResourceUnsubscribed []string                          `protobuf:"bytes,9,rep,name=resource_unsubscribed,json=resourceUnsubscribed,proto3" json:"resource_unsubscribed,omitempty"`
+	// instance this event originated from
+	Instance string `protobuf:"bytes,10,opt,name=instance,proto3" json:"instance,omitempty"`
+}
+
+func (x *EnvoyConfigurationEvent) Reset() {
+	*x = EnvoyConfigurationEvent{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_xds_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *EnvoyConfigurationEvent) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EnvoyConfigurationEvent) ProtoMessage() {}
+
+func (x *EnvoyConfigurationEvent) ProtoReflect() protoreflect.Message {
+	mi := &file_xds_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use EnvoyConfigurationEvent.ProtoReflect.Descriptor instead.
+func (*EnvoyConfigurationEvent) Descriptor() ([]byte, []int) {
+	return file_xds_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *EnvoyConfigurationEvent) GetTime() *timestamppb.Timestamp {
+	if x != nil {
+		return x.Time
+	}
+	return nil
+}
+
+func (x *EnvoyConfigurationEvent) GetMessage() string {
+	if x != nil {
+		return x.Message
+	}
+	return ""
+}
+
+func (x *EnvoyConfigurationEvent) GetCode() int32 {
+	if x != nil {
+		return x.Code
+	}
+	return 0
+}
+
+func (x *EnvoyConfigurationEvent) GetDetails() []*anypb.Any {
+	if x != nil {
+		return x.Details
+	}
+	return nil
+}
+
+func (x *EnvoyConfigurationEvent) GetConfigVersion() uint64 {
+	if x != nil {
+		return x.ConfigVersion
+	}
+	return 0
+}
+
+func (x *EnvoyConfigurationEvent) GetTypeUrl() string {
+	if x != nil {
+		return x.TypeUrl
+	}
+	return ""
+}
+
+func (x *EnvoyConfigurationEvent) GetKind() EnvoyConfigurationEvent_EventKind {
+	if x != nil {
+		return x.Kind
+	}
+	return EnvoyConfigurationEvent_EVENT_KIND_UNDEFINED
+}
+
+func (x *EnvoyConfigurationEvent) GetResourceSubscribed() []string {
+	if x != nil {
+		return x.ResourceSubscribed
+	}
+	return nil
+}
+
+func (x *EnvoyConfigurationEvent) GetResourceUnsubscribed() []string {
+	if x != nil {
+		return x.ResourceUnsubscribed
+	}
+	return nil
+}
+
+func (x *EnvoyConfigurationEvent) GetInstance() string {
+	if x != nil {
+		return x.Instance
+	}
+	return ""
+}
+
+var File_xds_proto protoreflect.FileDescriptor
+
+var file_xds_proto_rawDesc = []byte{
+	0x0a, 0x09, 0x78, 0x64, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0f, 0x70, 0x6f, 0x6d,
+	0x65, 0x72, 0x69, 0x75, 0x6d, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x1a, 0x1f, 0x67, 0x6f,
+	0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69,
+	0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x19, 0x67,
+	0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x61,
+	0x6e, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x5c, 0x0a, 0x18, 0x45, 0x6e, 0x76, 0x6f,
+	0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76,
+	0x65, 0x6e, 0x74, 0x73, 0x12, 0x40, 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x01,
+	0x20, 0x03, 0x28, 0x0b, 0x32, 0x28, 0x2e, 0x70, 0x6f, 0x6d, 0x65, 0x72, 0x69, 0x75, 0x6d, 0x2e,
+	0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x45, 0x6e, 0x76, 0x6f, 0x79, 0x43, 0x6f, 0x6e, 0x66,
+	0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x52, 0x06,
+	0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x22, 0x95, 0x04, 0x0a, 0x17, 0x45, 0x6e, 0x76, 0x6f, 0x79,
+	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65,
+	0x6e, 0x74, 0x12, 0x2e, 0x0a, 0x04, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
+	0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x04, 0x74, 0x69,
+	0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x12, 0x0a, 0x04,
+	0x63, 0x6f, 0x64, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x04, 0x63, 0x6f, 0x64, 0x65,
+	0x12, 0x2e, 0x0a, 0x07, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28,
+	0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x07, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73,
+	0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69,
+	0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67,
+	0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x0a, 0x08, 0x74, 0x79, 0x70, 0x65, 0x5f,
+	0x75, 0x72, 0x6c, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x74, 0x79, 0x70, 0x65, 0x55,
+	0x72, 0x6c, 0x12, 0x46, 0x0a, 0x04, 0x6b, 0x69, 0x6e, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0e,
+	0x32, 0x32, 0x2e, 0x70, 0x6f, 0x6d, 0x65, 0x72, 0x69, 0x75, 0x6d, 0x2e, 0x65, 0x76, 0x65, 0x6e,
+	0x74, 0x73, 0x2e, 0x45, 0x6e, 0x76, 0x6f, 0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x4b, 0x69, 0x6e, 0x64, 0x52, 0x04, 0x6b, 0x69, 0x6e, 0x64, 0x12, 0x2f, 0x0a, 0x13, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65,
+	0x64, 0x18, 0x08, 0x20, 0x03, 0x28, 0x09, 0x52, 0x12, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x64, 0x12, 0x33, 0x0a, 0x15, 0x72,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x75, 0x6e, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72,
+	0x69, 0x62, 0x65, 0x64, 0x18, 0x09, 0x20, 0x03, 0x28, 0x09, 0x52, 0x14, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x55, 0x6e, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x64,
+	0x12, 0x1a, 0x0a, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x18, 0x0a, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x22, 0x60, 0x0a, 0x09,
+	0x45, 0x76, 0x65, 0x6e, 0x74, 0x4b, 0x69, 0x6e, 0x64, 0x12, 0x18, 0x0a, 0x14, 0x45, 0x56, 0x45,
+	0x4e, 0x54, 0x5f, 0x4b, 0x49, 0x4e, 0x44, 0x5f, 0x55, 0x4e, 0x44, 0x45, 0x46, 0x49, 0x4e, 0x45,
+	0x44, 0x10, 0x00, 0x12, 0x1b, 0x0a, 0x17, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x44, 0x49, 0x53,
+	0x43, 0x4f, 0x56, 0x45, 0x52, 0x59, 0x5f, 0x52, 0x45, 0x51, 0x55, 0x45, 0x53, 0x54, 0x10, 0x01,
+	0x12, 0x1c, 0x0a, 0x18, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x44, 0x49, 0x53, 0x43, 0x4f, 0x56,
+	0x45, 0x52, 0x59, 0x5f, 0x52, 0x45, 0x53, 0x50, 0x4f, 0x4e, 0x53, 0x45, 0x10, 0x02, 0x42, 0x2e,
+	0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x6f, 0x6d,
+	0x65, 0x72, 0x69, 0x75, 0x6d, 0x2f, 0x70, 0x6f, 0x6d, 0x65, 0x72, 0x69, 0x75, 0x6d, 0x2f, 0x70,
+	0x6b, 0x67, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_xds_proto_rawDescOnce sync.Once
+	file_xds_proto_rawDescData = file_xds_proto_rawDesc
+)
+
+func file_xds_proto_rawDescGZIP() []byte {
+	file_xds_proto_rawDescOnce.Do(func() {
+		file_xds_proto_rawDescData = protoimpl.X.CompressGZIP(file_xds_proto_rawDescData)
+	})
+	return file_xds_proto_rawDescData
+}
+
+var file_xds_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_xds_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_xds_proto_goTypes = []interface{}{
+	(EnvoyConfigurationEvent_EventKind)(0), // 0: pomerium.events.EnvoyConfigurationEvent.EventKind
+	(*EnvoyConfigurationEvents)(nil),       // 1: pomerium.events.EnvoyConfigurationEvents
+	(*EnvoyConfigurationEvent)(nil),        // 2: pomerium.events.EnvoyConfigurationEvent
+	(*timestamppb.Timestamp)(nil),          // 3: google.protobuf.Timestamp
+	(*anypb.Any)(nil),                      // 4: google.protobuf.Any
+}
+var file_xds_proto_depIdxs = []int32{
+	2, // 0: pomerium.events.EnvoyConfigurationEvents.values:type_name -> pomerium.events.EnvoyConfigurationEvent
+	3, // 1: pomerium.events.EnvoyConfigurationEvent.time:type_name -> google.protobuf.Timestamp
+	4, // 2: pomerium.events.EnvoyConfigurationEvent.details:type_name -> google.protobuf.Any
+	0, // 3: pomerium.events.EnvoyConfigurationEvent.kind:type_name -> pomerium.events.EnvoyConfigurationEvent.EventKind
+	4, // [4:4] is the sub-list for method output_type
+	4, // [4:4] is the sub-list for method input_type
+	4, // [4:4] is the sub-list for extension type_name
+	4, // [4:4] is the sub-list for extension extendee
+	0, // [0:4] is the sub-list for field type_name
+}
+
+func init() { file_xds_proto_init() }
+func file_xds_proto_init() {
+	if File_xds_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_xds_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*EnvoyConfigurationEvents); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_xds_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*EnvoyConfigurationEvent); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_xds_proto_rawDesc,
+			NumEnums:      1,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_xds_proto_goTypes,
+		DependencyIndexes: file_xds_proto_depIdxs,
+		EnumInfos:         file_xds_proto_enumTypes,
+		MessageInfos:      file_xds_proto_msgTypes,
+	}.Build()
+	File_xds_proto = out.File
+	file_xds_proto_rawDesc = nil
+	file_xds_proto_goTypes = nil
+	file_xds_proto_depIdxs = nil
+}

pkg/grpc/events/xds.proto

@@ -0,0 +1,36 @@
+syntax = "proto3";
+
+package pomerium.events;
+option go_package = "github.com/pomerium/pomerium/pkg/grpc/events";
+
+import "google/protobuf/timestamp.proto";
+import "google/protobuf/any.proto";
+
+// EnvoyConfigurationEvents is a list of envoy configuration events.
+message EnvoyConfigurationEvents {
+  repeated EnvoyConfigurationEvent values = 1;
+}
+
+// EnvoyConfigurationEvent is an envoy configuration event.
+message EnvoyConfigurationEvent {
+  google.protobuf.Timestamp time = 1;
+  string message = 2;
+  int32 code = 3;
+  repeated google.protobuf.Any details = 4;
+  // databroker config version
+  uint64 config_version = 5;
+  // envoy resource type (i.e. listener, cluster)
+  string type_url = 6;
+  enum EventKind {
+    EVENT_KIND_UNDEFINED = 0;
+    // envoy_service_discovery_v3.DeltaDiscoveryRequest
+    EVENT_DISCOVERY_REQUEST = 1;
+    // envoy_service_discovery_v3.DeltaDiscoveryResponse
+    EVENT_DISCOVERY_RESPONSE = 2;
+  }
+  EventKind kind = 7;
+  repeated string resource_subscribed = 8;
+  repeated string resource_unsubscribed = 9;
+  // instance this event originated from
+  string instance = 10;
+}

pkg/grpc/protoc.bash

@@ -97,3 +97,7 @@ _import_paths=$(join_by , "${_imports[@]}")
 ../../scripts/protoc -I ./user/ \
   --go_out="$_import_paths,plugins=grpc,paths=source_relative:./user/." \
   ./user/user.proto
+
+../../scripts/protoc -I ./events/ -I ./ \
+  --go_out="$_import_paths,plugins=grpc,paths=source_relative:./events/." \
+  ./events/xds.proto