Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions pkg/agent/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -280,18 +280,16 @@ func (a *Agent) Run(ctx context.Context) error {
storeService.Run,
catalog.ReconfigureTask(a.c.Log.WithField(telemetry.SubsystemName, "reconfigurer"), cat),
}
var apiReadyChannels []chan struct{}

if a.c.BindAddress != nil {
agentEndpoints := a.newEndpoints(metrics, mgr, workloadAttestor)
go func() {
agentEndpoints.WaitForListening(readyForHealthChecks)
a.started = true
}()
listening := make(chan struct{})
apiReadyChannels = append(apiReadyChannels, listening)
go agentEndpoints.WaitForListening(listening)
tasks = append(tasks, agentEndpoints.ListenAndServe)
} else {
a.c.Log.WithField("apis", "Workload and SDS APIs").Info("Skipping agent APIs because public endpoint is disabled")
a.started = true
close(readyForHealthChecks)
}

if a.c.AdminBindAddress != nil {
Expand All @@ -314,9 +312,20 @@ func (a *Agent) Run(ctx context.Context) error {
if err != nil {
return fmt.Errorf("failed to create broker endpoints: %w", err)
}
listening := make(chan struct{})
apiReadyChannels = append(apiReadyChannels, listening)
go brokerEndpoints.WaitForListening(listening)
tasks = append(tasks, brokerEndpoints.ListenAndServe)
}

go func() {
for _, listening := range apiReadyChannels {
<-listening
}
a.started = true
close(readyForHealthChecks)
}()

if a.c.LogReopener != nil {
tasks = append(tasks, a.c.LogReopener)
}
Expand Down
17 changes: 17 additions & 0 deletions pkg/agent/broker/endpoints.go
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,10 @@ type AllowedReferenceType struct {

type Endpoints struct {
c *Config

hooks struct {
listening chan struct{}
}
}

func New(c *Config) (*Endpoints, error) {
Expand All @@ -107,9 +111,21 @@ func New(c *Config) (*Endpoints, error) {
}
return &Endpoints{
c: c,
hooks: struct {
listening chan struct{}
}{
listening: make(chan struct{}),
},
}, nil
}

// WaitForListening signals on listening after all configured Broker API
// listeners have been created.
func (e *Endpoints) WaitForListening(listening chan struct{}) {
<-e.hooks.listening
listening <- struct{}{}
}

func (e *Endpoints) ListenAndServe(ctx context.Context) error {
unaryInterceptor, streamInterceptor := middleware.Interceptors(
middleware.Chain(
Expand Down Expand Up @@ -178,6 +194,7 @@ func (e *Endpoints) ListenAndServe(ctx context.Context) error {
telemetry.Address: l.Addr().String(),
}).Info("Starting SPIFFE Broker Endpoint")
}
close(e.hooks.listening)
Comment thread
matheuscscp marked this conversation as resolved.
Outdated

// Fan one gRPC server out across every listener with an errgroup. The
// first goroutine to error (or context cancellation) cancels the
Expand Down
Loading