diff --git a/pkg/collector/managers/cluster/assigner.go b/pkg/collector/managers/cluster/assigner.go index a0ac5170..548c8934 100644 --- a/pkg/collector/managers/cluster/assigner.go +++ b/pkg/collector/managers/cluster/assigner.go @@ -41,13 +41,19 @@ type restAssigner struct { logger *slog.Logger } -func NewAssigner(store *collstore.Store) Assigner { +// NewAssigner returns an Assigner that pushes assignments to cluster members. +// client is the HTTP client built from clustering.tls (see +// newClusteringHTTPClient); if nil, a plain client is used as a fallback. +func NewAssigner(store *collstore.Store, client *http.Client) Assigner { + if client == nil { + client = &http.Client{ + Timeout: 10 * time.Second, + } + } return &restAssigner{ store: store, logger: slog.With("component", "assignment-pusher"), - client: &http.Client{ - Timeout: 10 * time.Second, - }, + client: client, } } diff --git a/pkg/collector/managers/cluster/cluster_integration_test.go b/pkg/collector/managers/cluster/cluster_integration_test.go index b65be7fd..9f546c6f 100644 --- a/pkg/collector/managers/cluster/cluster_integration_test.go +++ b/pkg/collector/managers/cluster/cluster_integration_test.go @@ -67,7 +67,7 @@ func TestRestAssigner_postsAssignments(t *testing.T) { t.Cleanup(srv.Close) st := testutil.NewTestStore(t) - assigner := NewAssigner(st).(*restAssigner) + assigner := NewAssigner(st, nil).(*restAssigner) assigner.client = srv.Client() addr := strings.TrimPrefix(srv.URL, "http://") diff --git a/pkg/collector/managers/cluster/cluster_manager.go b/pkg/collector/managers/cluster/cluster_manager.go index 8df80fe9..530db310 100644 --- a/pkg/collector/managers/cluster/cluster_manager.go +++ b/pkg/collector/managers/cluster/cluster_manager.go @@ -2,6 +2,7 @@ package cluster_manager import ( "context" + "crypto/tls" "errors" "fmt" "log/slog" @@ -17,6 +18,7 @@ import ( "golang.org/x/sync/semaphore" + "github.com/openconfig/gnmic/pkg/api/utils" apiconst "github.com/openconfig/gnmic/pkg/collector/api/const" "github.com/openconfig/gnmic/pkg/collector/env" collstore "github.com/openconfig/gnmic/pkg/collector/store" @@ -26,9 +28,43 @@ import ( "github.com/zestor-dev/zestor/store" ) +// newClusteringHTTPClient builds the HTTP client used for leader->member API +// calls (target assignment/unassignment/deletion), honoring clustering.tls. +// It mirrors (*App).createAPIClient in pkg/app/clustering.go so that collector +// mode has the same inter-member TLS behavior as subscribe mode. +func newClusteringHTTPClient(cfg *config.Clustering) (*http.Client, error) { + if cfg == nil || cfg.TLS == nil { + return &http.Client{ + Timeout: defaultClusteringHTTPTimeout, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, + }, + }, + }, nil + } + tlsConfig, err := utils.NewTLSConfig( + cfg.TLS.CaFile, + cfg.TLS.CertFile, + cfg.TLS.KeyFile, "", + cfg.TLS.SkipVerify, + false) + if err != nil { + return nil, err + } + return &http.Client{ + Timeout: defaultClusteringHTTPTimeout, + Transport: &http.Transport{ + TLSClientConfig: tlsConfig, + }, + }, nil +} + const ( protocolTagName = "__protocol" retryRegistrationBackoff = 2 * time.Second + + defaultClusteringHTTPTimeout = 10 * time.Second ) type ClusterManager struct { @@ -66,7 +102,8 @@ func NewClusterManager(store *collstore.Store) *ClusterManager { locker: nil, lockCheckLimiter: make(chan struct{}, 64), // TODO: make this configurable rebalancingSem: semaphore.NewWeighted(1), - apiClient: &http.Client{Timeout: 10 * time.Second}, // TODO: + // replaced in Start() with a client built from clustering.tls + apiClient: &http.Client{Timeout: defaultClusteringHTTPTimeout}, } } @@ -94,6 +131,13 @@ func (c *ClusterManager) Start(ctx context.Context, locker lockers.Locker, wg *s c.clusteringConfig = clustering env.ExpandClusterEnv(c.clusteringConfig) + // build the inter-member API client from clustering.tls so that + // leader->member calls work over a TLS-secured API (private CA / mTLS). + c.apiClient, err = newClusteringHTTPClient(c.clusteringConfig) + if err != nil { + return fmt.Errorf("failed to build clustering API client: %w", err) + } + apiConfig, ok, err := c.store.Config.Get("api-server", "api-server") if err != nil { return err @@ -118,7 +162,7 @@ func (c *ClusterManager) Start(ctx context.Context, locker lockers.Locker, wg *s return err } c.membership = NewMembership(c.locker, clustering, c.logger) - c.assigner = NewAssigner(c.store) + c.assigner = NewAssigner(c.store, c.apiClient) // start registration to register the api service wg.Add(1)