Use SubscribeToLocalBundle UpstreamAuthority RPC if available

Signed-off-by: Sorin Dumitru <[email protected]>

Author: sorindumitru

pkg/server/ca/manager/manager.go

@@ -71,6 +71,7 @@ type AuthorityManager interface {
 	IsUpstreamAuthority() bool
 	PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error)
 	NotifyTaintedX509Authority(ctx context.Context, authorityID string) error
+	SubscribeToLocalBundle(ctx context.Context) error
 }
 
 type Config struct {
@@ -434,6 +435,7 @@ func (m *Manager) PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) (
 			})
 		case err != nil:
 			return nil, err
+
 		default:
 			return upstreamJWTKeys, nil
 		}
@@ -447,6 +449,22 @@ func (m *Manager) PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) (
 	return bundle.JwtSigningKeys, nil
 }
 
+func (m *Manager) SubscribeToLocalBundle(ctx context.Context) error {
+	if m.upstreamClient == nil {
+		return nil
+	}
+
+	err := m.upstreamClient.SubscribeToLocalBundle(ctx)
+	switch {
+	case status.Code(err) == codes.Unimplemented:
+		return nil
+	case err != nil:
+		return err
+	default:
+		return nil
+	}
+}
+
 func (m *Manager) PruneBundle(ctx context.Context) (err error) {
 	counter := telemetry_server.StartCAManagerPruneBundleCall(m.c.Metrics)
 	defer counter.Done(&err)

pkg/server/ca/rotator/rotator.go

@@ -37,6 +37,8 @@ type CAManager interface {
 	ActivateJWTKey(ctx context.Context)
 	RotateJWTKey(ctx context.Context)
 
+	SubscribeToLocalBundle(ctx context.Context) error
+
 	PruneBundle(ctx context.Context) error
 	PruneCAJournals(ctx context.Context) error
 }
@@ -77,10 +79,25 @@ func (r *Rotator) Run(ctx context.Context) error {
 	if err := r.c.Manager.NotifyBundleLoaded(ctx); err != nil {
 		return err
 	}
+
 	err := util.RunTasks(ctx,
 		func(ctx context.Context) error {
 			return r.rotateEvery(ctx, rotateInterval)
 		},
+		func(ctx context.Context) error {
+			var lastError error
+			for {
+				select {
+				case <-ctx.Done():
+					return lastError
+				case <-time.After(5 * time.Second):
+					lastError = r.c.Manager.SubscribeToLocalBundle(ctx)
+					if lastError == nil {
+						return nil
+					}
+				}
+			}
+		},
 		func(ctx context.Context) error {
 			return r.pruneBundleEvery(ctx, pruneBundleInterval)
 		},

pkg/server/ca/rotator/rotator_test.go

@@ -506,6 +506,10 @@ func (f *fakeCAManager) RotateJWTKey(context.Context) {
 	f.jwtKeyCh <- struct{}{}
 }
 
+func (f *fakeCAManager) SubscribeToLocalBundle(ctx context.Context) error {
+	return nil
+}
+
 func (f *fakeCAManager) PruneBundle(context.Context) error {
 	defer func() {
 		f.pruneBundleCh <- struct{}{}