manager.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546
  1. /*
  2. Copyright © The ESO Authors
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. https://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. // Package clientmanager provides a Manager for provider clients
  14. package clientmanager
  15. import (
  16. "context"
  17. "errors"
  18. "fmt"
  19. "regexp"
  20. "strings"
  21. "sync"
  22. "github.com/go-logr/logr"
  23. v1 "k8s.io/api/core/v1"
  24. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  25. "k8s.io/apimachinery/pkg/labels"
  26. "k8s.io/apimachinery/pkg/types"
  27. ctrl "sigs.k8s.io/controller-runtime"
  28. "sigs.k8s.io/controller-runtime/pkg/client"
  29. esv1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1"
  30. esv1alpha1 "github.com/external-secrets/external-secrets/apis/externalsecrets/v1alpha1"
  31. adapterstore "github.com/external-secrets/external-secrets/providers/v2/adapter/store"
  32. "github.com/external-secrets/external-secrets/providers/v2/common/grpc"
  33. )
  34. const (
  35. errGetClusterSecretStore = "could not get ClusterSecretStore %q, %w"
  36. errGetSecretStore = "could not get SecretStore %q, %w"
  37. errSecretStoreNotReady = "%s %q is not ready"
  38. errClusterStoreMismatch = "using cluster store %q is not allowed from namespace %q: denied by spec.condition"
  39. providerMetricsLabel = "provider"
  40. clusterProviderMetricsLabel = "cluster-provider"
  41. cacheInvalidationGeneration = "generation_change"
  42. cacheInvalidationMismatch = "store_mismatch"
  43. runtimeRefCacheKeyType = "runtime-ref"
  44. errRuntimeRefProviderClassClusterStore = "ClusterSecretStore runtimeRef.kind must not be \"ProviderClass\""
  45. )
  46. var (
  47. // globalV2ConnectionPool is a singleton connection pool for v2 gRPC providers.
  48. // It persists across all reconciles and Manager instances to enable connection reuse.
  49. // Initialized once on first use and shared globally.
  50. globalV2ConnectionPool *grpc.ConnectionPool
  51. globalV2ConnectionPoolOnce sync.Once
  52. globalV2ConnectionPoolLog logr.Logger
  53. )
  54. // initGlobalV2ConnectionPool initializes the global connection pool for v2 providers.
  55. // This is called once on first use via sync.Once.
  56. func initGlobalV2ConnectionPool() {
  57. globalV2ConnectionPoolLog = ctrl.Log.WithName("v2-connection-pool")
  58. poolConfig := grpc.DefaultPoolConfig()
  59. globalV2ConnectionPool = grpc.NewConnectionPool(poolConfig)
  60. globalV2ConnectionPoolLog.Info("global v2 connection pool initialized",
  61. "maxIdleTime", poolConfig.MaxIdleTime.String(),
  62. "maxLifetime", poolConfig.MaxLifetime.String(),
  63. "healthCheckInterval", poolConfig.HealthCheckInterval.String())
  64. }
  65. // getGlobalV2ConnectionPool returns the global connection pool, initializing it if needed.
  66. func getGlobalV2ConnectionPool() *grpc.ConnectionPool {
  67. globalV2ConnectionPoolOnce.Do(initGlobalV2ConnectionPool)
  68. return globalV2ConnectionPool
  69. }
  70. // v2PooledConnection tracks connection info needed to release connections back to the pool.
  71. type v2PooledConnection struct {
  72. address string
  73. tlsConfig *grpc.TLSConfig
  74. }
  75. // Manager stores instances of provider clients
  76. // At any given time we must have no more than one instance
  77. // of a client (due to limitations in GCP / see mutexlock there)
  78. // If the controller requests another instance of a given client
  79. // we will close the old client first and then construct a new one.
  80. type Manager struct {
  81. log logr.Logger
  82. client client.Client
  83. controllerClass string
  84. enableFloodgate bool
  85. // store clients by provider type
  86. clientMap map[clientKey]*clientVal
  87. // Track v2 provider connections for release back to pool
  88. v2PooledConnections []v2PooledConnection
  89. }
  90. type clientKey struct {
  91. providerType string
  92. // For v2 providers, store the provider name and namespace
  93. v2ProviderName string
  94. v2ProviderNamespace string
  95. runtimeSourceNamespace string
  96. }
  97. type clientVal struct {
  98. client esv1.SecretsClient
  99. store esv1.GenericStore
  100. }
  101. func providerMetricsLabelForKey(key clientKey) string {
  102. if key.v2ProviderName == "" {
  103. return "unknown"
  104. }
  105. if key.v2ProviderNamespace == "" {
  106. return clusterProviderMetricsLabel
  107. }
  108. return providerMetricsLabel
  109. }
  110. // NewManager constructs a new manager with defaults.
  111. func NewManager(ctrlClient client.Client, controllerClass string, enableFloodgate bool) *Manager {
  112. log := ctrl.Log.WithName("clientmanager")
  113. return &Manager{
  114. log: log,
  115. client: ctrlClient,
  116. controllerClass: controllerClass,
  117. enableFloodgate: enableFloodgate,
  118. clientMap: make(map[clientKey]*clientVal),
  119. }
  120. }
  121. // GetFromStore returns a provider client from the given store.
  122. // Do not close the client returned from this func, instead close
  123. // the manager once you're done with reconciling the external secret.
  124. func (m *Manager) GetFromStore(ctx context.Context, store esv1.GenericStore, namespace string) (esv1.SecretsClient, error) {
  125. if store.GetSpec().RuntimeRef != nil {
  126. return m.getRuntimeRefClient(ctx, store, namespace)
  127. }
  128. storeProvider, err := esv1.GetProvider(store)
  129. if err != nil {
  130. return nil, err
  131. }
  132. secretClient := m.getStoredClient(ctx, storeProvider, store)
  133. if secretClient != nil {
  134. return secretClient, nil
  135. }
  136. m.log.V(1).Info("creating new client",
  137. "provider", fmt.Sprintf("%T", storeProvider),
  138. "store", fmt.Sprintf("%s/%s", store.GetNamespace(), store.GetName()))
  139. // secret client is created only if we are going to refresh
  140. // this skip an unnecessary check/request in the case we are not going to do anything
  141. secretClient, err = storeProvider.NewClient(ctx, store, m.client, namespace)
  142. if err != nil {
  143. return nil, err
  144. }
  145. idx := storeKey(storeProvider)
  146. m.clientMap[idx] = &clientVal{
  147. client: secretClient,
  148. store: store,
  149. }
  150. return secretClient, nil
  151. }
  152. func (m *Manager) getRuntimeRefClient(ctx context.Context, store esv1.GenericStore, namespace string) (esv1.SecretsClient, error) {
  153. runtimeRef := store.GetSpec().RuntimeRef
  154. cacheKey := runtimeRefStoreKey(store, namespace)
  155. if cached := m.getStoredRuntimeRefClient(ctx, cacheKey, store); cached != nil {
  156. return cached, nil
  157. }
  158. runtimeDetails, err := m.resolveRuntimeRef(ctx, store, runtimeRef)
  159. if err != nil {
  160. return nil, err
  161. }
  162. providerRef, err := buildProviderReference(store, namespace)
  163. if err != nil {
  164. return nil, err
  165. }
  166. tlsSecretNamespace := grpc.ResolveTLSSecretNamespace(runtimeDetails.address, "", "", "")
  167. tlsConfig, err := grpc.LoadClientTLSConfig(ctx, m.client, runtimeDetails.address, tlsSecretNamespace)
  168. if err != nil {
  169. return nil, fmt.Errorf("failed to load TLS config for %s %q: %w", runtimeDetails.kind, runtimeRef.Name, err)
  170. }
  171. pool := getGlobalV2ConnectionPool()
  172. grpcClient, err := pool.Get(ctx, runtimeDetails.address, tlsConfig)
  173. if err != nil {
  174. return nil, fmt.Errorf("failed to get gRPC client from pool for %s %q: %w", runtimeDetails.kind, runtimeRef.Name, err)
  175. }
  176. providerClient := adapterstore.NewClientWithCloser(grpcClient, providerRef, namespace, func(context.Context) error {
  177. pool.Release(runtimeDetails.address, tlsConfig)
  178. return nil
  179. })
  180. m.clientMap[cacheKey] = &clientVal{
  181. client: providerClient,
  182. store: store,
  183. }
  184. return providerClient, nil
  185. }
  186. type runtimeRefDetails struct {
  187. kind string
  188. address string
  189. }
  190. func (m *Manager) resolveRuntimeRef(ctx context.Context, store esv1.GenericStore, runtimeRef *esv1.StoreRuntimeRef) (*runtimeRefDetails, error) {
  191. runtimeKind, err := runtimeRefKindForStore(store, runtimeRef)
  192. if err != nil {
  193. return nil, err
  194. }
  195. switch runtimeKind {
  196. case esv1.StoreRuntimeRefKindProviderClass:
  197. var runtimeClass esv1alpha1.ProviderClass
  198. if err := m.client.Get(ctx, types.NamespacedName{Name: runtimeRef.Name, Namespace: store.GetNamespace()}, &runtimeClass); err != nil {
  199. return nil, fmt.Errorf("failed to get %s %q: %w", runtimeKind, runtimeRef.Name, err)
  200. }
  201. return &runtimeRefDetails{kind: runtimeKind, address: runtimeClass.Spec.Address}, nil
  202. case esv1.StoreRuntimeRefKindClusterProviderClass:
  203. var runtimeClass esv1alpha1.ClusterProviderClass
  204. if err := m.client.Get(ctx, types.NamespacedName{Name: runtimeRef.Name}, &runtimeClass); err != nil {
  205. return nil, fmt.Errorf("failed to get %s %q: %w", runtimeKind, runtimeRef.Name, err)
  206. }
  207. return &runtimeRefDetails{kind: runtimeKind, address: runtimeClass.Spec.Address}, nil
  208. default:
  209. return nil, fmt.Errorf("unsupported runtimeRef kind %q", runtimeKind)
  210. }
  211. }
  212. func runtimeRefKindForStore(store esv1.GenericStore, runtimeRef *esv1.StoreRuntimeRef) (string, error) {
  213. runtimeKind := runtimeRef.Kind
  214. if runtimeKind == "" {
  215. if store.GetKind() == esv1.ClusterSecretStoreKind {
  216. return esv1.StoreRuntimeRefKindClusterProviderClass, nil
  217. }
  218. return esv1.StoreRuntimeRefKindProviderClass, nil
  219. }
  220. if store.GetKind() == esv1.ClusterSecretStoreKind && runtimeKind == esv1.StoreRuntimeRefKindProviderClass {
  221. return "", fmt.Errorf(errRuntimeRefProviderClassClusterStore)
  222. }
  223. return runtimeKind, nil
  224. }
  225. func runtimeRefStoreKey(store esv1.GenericStore, sourceNamespace string) clientKey {
  226. return clientKey{
  227. providerType: runtimeRefCacheKeyType + ":" + store.GetKind(),
  228. v2ProviderName: store.GetName(),
  229. v2ProviderNamespace: store.GetNamespace(),
  230. runtimeSourceNamespace: sourceNamespace,
  231. }
  232. }
  233. func (m *Manager) getStoredRuntimeRefClient(ctx context.Context, key clientKey, store esv1.GenericStore) esv1.SecretsClient {
  234. val, ok := m.clientMap[key]
  235. if !ok {
  236. return nil
  237. }
  238. valGVK, err := m.client.GroupVersionKindFor(val.store)
  239. if err != nil {
  240. return nil
  241. }
  242. storeGVK, err := m.client.GroupVersionKindFor(store)
  243. if err != nil {
  244. return nil
  245. }
  246. if val.store.GetObjectMeta().Generation == store.GetGeneration() &&
  247. valGVK == storeGVK &&
  248. val.store.GetName() == store.GetName() &&
  249. val.store.GetNamespace() == store.GetNamespace() {
  250. clientManagerMetrics.RecordCacheHit(providerMetricsLabelForKey(key))
  251. return val.client
  252. }
  253. _ = val.client.Close(ctx)
  254. delete(m.clientMap, key)
  255. reason := cacheInvalidationMismatch
  256. if val.store.GetObjectMeta().Generation != store.GetGeneration() {
  257. reason = cacheInvalidationGeneration
  258. }
  259. clientManagerMetrics.RecordCacheInvalidation(providerMetricsLabelForKey(key), reason)
  260. return nil
  261. }
  262. // Get returns a provider client from the given storeRef or sourceRef.secretStoreRef
  263. // while sourceRef.SecretStoreRef takes precedence over storeRef.
  264. // Do not close the client returned from this func, instead close
  265. // the manager once you're done with recinciling the external secret.
  266. func (m *Manager) Get(ctx context.Context, storeRef esv1.SecretStoreRef, namespace string, sourceRef *esv1.StoreGeneratorSourceRef) (esv1.SecretsClient, error) {
  267. if sourceRef != nil && sourceRef.SecretStoreRef != nil {
  268. storeRef = *sourceRef.SecretStoreRef
  269. }
  270. store, err := m.getStore(ctx, &storeRef, namespace)
  271. if err != nil {
  272. return nil, err
  273. }
  274. // check if store should be handled by this controller instance
  275. if !ShouldProcessStore(store, m.controllerClass) {
  276. return nil, errors.New("can not reference unmanaged store")
  277. }
  278. // when using ClusterSecretStore, validate the ClusterSecretStore namespace conditions
  279. shouldProcess, err := m.shouldProcessSecret(store, namespace)
  280. if err != nil {
  281. return nil, err
  282. }
  283. if !shouldProcess {
  284. return nil, fmt.Errorf(errClusterStoreMismatch, store.GetName(), namespace)
  285. }
  286. if m.enableFloodgate {
  287. err := assertStoreIsUsable(store)
  288. if err != nil {
  289. return nil, err
  290. }
  291. }
  292. return m.GetFromStore(ctx, store, namespace)
  293. }
  294. // returns a previously stored client from the cache if store and store-version match
  295. // if a client exists for the same provider which points to a different store or store version
  296. // it will be cleaned up.
  297. func (m *Manager) getStoredClient(ctx context.Context, storeProvider esv1.ProviderInterface, store esv1.GenericStore) esv1.SecretsClient {
  298. idx := storeKey(storeProvider)
  299. val, ok := m.clientMap[idx]
  300. if !ok {
  301. return nil
  302. }
  303. valGVK, err := m.client.GroupVersionKindFor(val.store)
  304. if err != nil {
  305. return nil
  306. }
  307. storeGVK, err := m.client.GroupVersionKindFor(store)
  308. if err != nil {
  309. return nil
  310. }
  311. storeName := fmt.Sprintf("%s/%s", store.GetNamespace(), store.GetName())
  312. // return client if it points to the very same store
  313. if val.store.GetObjectMeta().Generation == store.GetGeneration() &&
  314. valGVK == storeGVK &&
  315. val.store.GetName() == store.GetName() &&
  316. val.store.GetNamespace() == store.GetNamespace() {
  317. m.log.V(1).Info("reusing stored client",
  318. "provider", fmt.Sprintf("%T", storeProvider),
  319. "store", storeName)
  320. // Record cache hit
  321. clientManagerMetrics.RecordCacheHit(providerMetricsLabelForKey(idx))
  322. return val.client
  323. }
  324. m.log.V(1).Info("cleaning up client",
  325. "provider", fmt.Sprintf("%T", storeProvider),
  326. "store", storeName)
  327. // if we have a client, but it points to a different store
  328. // we must clean it up
  329. _ = val.client.Close(ctx)
  330. delete(m.clientMap, idx)
  331. // Record cache invalidation
  332. providerType := providerMetricsLabelForKey(idx)
  333. reason := cacheInvalidationMismatch
  334. if idx.v2ProviderName != "" {
  335. if val.store.GetObjectMeta().Generation != store.GetGeneration() {
  336. reason = cacheInvalidationGeneration
  337. }
  338. }
  339. clientManagerMetrics.RecordCacheInvalidation(providerType, reason)
  340. return nil
  341. }
  342. func storeKey(storeProvider esv1.ProviderInterface) clientKey {
  343. return clientKey{
  344. providerType: fmt.Sprintf("%T", storeProvider),
  345. }
  346. }
  347. // getStore fetches the (Cluster)SecretStore from the kube-apiserver
  348. // and returns a GenericStore representing it.
  349. func (m *Manager) getStore(ctx context.Context, storeRef *esv1.SecretStoreRef, namespace string) (esv1.GenericStore, error) {
  350. ref := types.NamespacedName{
  351. Name: storeRef.Name,
  352. }
  353. if storeRef.Kind == esv1.ClusterSecretStoreKind {
  354. var store esv1.ClusterSecretStore
  355. err := m.client.Get(ctx, ref, &store)
  356. if err != nil {
  357. return nil, fmt.Errorf(errGetClusterSecretStore, ref.Name, err)
  358. }
  359. return &store, nil
  360. }
  361. ref.Namespace = namespace
  362. var store esv1.SecretStore
  363. err := m.client.Get(ctx, ref, &store)
  364. if err != nil {
  365. return nil, fmt.Errorf(errGetSecretStore, ref.Name, err)
  366. }
  367. return &store, nil
  368. }
  369. // Close cleans up all clients.
  370. // For v1 providers, it closes the clients directly.
  371. // For v2 providers, it releases connections back to the pool for reuse.
  372. func (m *Manager) Close(ctx context.Context) error {
  373. var errs []string
  374. // Release v2 pooled connections back to the pool
  375. pool := getGlobalV2ConnectionPool()
  376. for _, pooledConn := range m.v2PooledConnections {
  377. pool.Release(pooledConn.address, pooledConn.tlsConfig)
  378. m.log.V(1).Info("released v2 connection back to pool",
  379. "address", pooledConn.address)
  380. }
  381. m.v2PooledConnections = nil
  382. // Close cached clients. Runtime-ref-backed clients release their pooled
  383. // connection through their Close implementation.
  384. for key, val := range m.clientMap {
  385. err := val.client.Close(ctx)
  386. if err != nil {
  387. errs = append(errs, err.Error())
  388. }
  389. delete(m.clientMap, key)
  390. }
  391. if len(errs) != 0 {
  392. return fmt.Errorf("errors while closing clients: %s", strings.Join(errs, ", "))
  393. }
  394. return nil
  395. }
  396. // validateNamespaceConditions checks if a namespace matches the given conditions.
  397. // Returns true if the namespace is allowed, false if denied.
  398. func (m *Manager) validateNamespaceConditions(conditions []esv1.ClusterSecretStoreCondition, ns string) (bool, error) {
  399. if len(conditions) == 0 {
  400. return true, nil
  401. }
  402. namespace := v1.Namespace{}
  403. if err := m.client.Get(context.Background(), client.ObjectKey{Name: ns}, &namespace); err != nil {
  404. return false, fmt.Errorf("failed to get a namespace %q: %w", ns, err)
  405. }
  406. nsLabels := labels.Set(namespace.GetLabels())
  407. for _, condition := range conditions {
  408. var labelSelectors []*metav1.LabelSelector
  409. if condition.NamespaceSelector != nil {
  410. labelSelectors = append(labelSelectors, condition.NamespaceSelector)
  411. }
  412. for _, n := range condition.Namespaces {
  413. labelSelectors = append(labelSelectors, &metav1.LabelSelector{
  414. MatchLabels: map[string]string{
  415. "kubernetes.io/metadata.name": n,
  416. },
  417. })
  418. }
  419. for _, ls := range labelSelectors {
  420. selector, err := metav1.LabelSelectorAsSelector(ls)
  421. if err != nil {
  422. return false, fmt.Errorf("failed to convert label selector into selector %v: %w", ls, err)
  423. }
  424. if selector.Matches(nsLabels) {
  425. return true, nil
  426. }
  427. }
  428. for _, reg := range condition.NamespaceRegexes {
  429. match, err := regexp.MatchString(reg, ns)
  430. if err != nil {
  431. // Should not happen since store validation already verified the regexes.
  432. return false, fmt.Errorf("failed to compile regex %v: %w", reg, err)
  433. }
  434. if match {
  435. return true, nil
  436. }
  437. }
  438. }
  439. return false, nil
  440. }
  441. // shouldProcessSecret validates if a secret should be processed based on namespace conditions.
  442. // This is a wrapper around validateNamespaceConditions for backward compatibility with GenericStore.
  443. func (m *Manager) shouldProcessSecret(store esv1.GenericStore, ns string) (bool, error) {
  444. // Only check conditions for cluster-scoped resources.
  445. if store.GetKind() != esv1.ClusterSecretStoreKind {
  446. return true, nil
  447. }
  448. return m.validateNamespaceConditions(store.GetSpec().Conditions, ns)
  449. }
  450. // assertStoreIsUsable asserts that the store is ready to use.
  451. func assertStoreIsUsable(store esv1.GenericStore) error {
  452. if store == nil {
  453. return nil
  454. }
  455. condition := GetSecretStoreCondition(store.GetStatus(), esv1.SecretStoreReady)
  456. if condition == nil || condition.Status != v1.ConditionTrue {
  457. return fmt.Errorf(errSecretStoreNotReady, store.GetKind(), store.GetName())
  458. }
  459. return nil
  460. }
  461. // ShouldProcessStore returns true if the store should be processed.
  462. func ShouldProcessStore(store esv1.GenericStore, class string) bool {
  463. if store == nil || store.GetSpec().Controller == "" || store.GetSpec().Controller == class {
  464. return true
  465. }
  466. return false
  467. }
  468. // GetSecretStoreCondition returns the condition with the provided type.
  469. func GetSecretStoreCondition(status esv1.SecretStoreStatus, condType esv1.SecretStoreConditionType) *esv1.SecretStoreStatusCondition {
  470. for i := range status.Conditions {
  471. c := status.Conditions[i]
  472. if c.Type == condType {
  473. return &c
  474. }
  475. }
  476. return nil
  477. }