package helm import ( "context" "sync" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "github.com/sophotechlabs/spinoza/internal/api" "github.com/sophotechlabs/spinoza/internal/safe" ) const readConcurrency = 2 type decoded struct { version string release api.HelmRelease } type releaseCache struct { mu sync.Mutex items map[string]decoded } func newReleaseCache() *releaseCache { return &releaseCache{items: map[string]decoded{}} } func cacheKey(ref storedRef) string { return ref.driver + " " + ref.namespace + "/" + ref.object } func (c *releaseCache) get(ref storedRef) (api.HelmRelease, bool) { c.mu.Lock() defer c.mu.Unlock() held, ok := c.items[cacheKey(ref)] if !ok || held.version != ref.version { return api.HelmRelease{}, false } return held.release, true } func (c *releaseCache) put(ref storedRef, release api.HelmRelease) { c.mu.Lock() defer c.mu.Unlock() c.items[cacheKey(ref)] = decoded{version: ref.version, release: release} } func (c *releaseCache) keep(refs []storedRef) { live := make(map[string]struct{}, len(refs)) for _, ref := range refs { live[cacheKey(ref)] = struct{}{} } c.mu.Lock() defer c.mu.Unlock() for key := range c.items { _, wanted := live[key] if !wanted { delete(c.items, key) } } } func (s *Service) read(ctx context.Context, refs []storedRef) ([]api.HelmRelease, int) { out, readable := s.readWithStatus(ctx, refs) undecodable := 0 for _, ok := range readable { if !ok { undecodable++ } } return out, undecodable } func (s *Service) readWithStatus(ctx context.Context, refs []storedRef) ([]api.HelmRelease, []bool) { out := make([]api.HelmRelease, len(refs)) readable := make([]bool, len(refs)) var mu sync.Mutex var group sync.WaitGroup slots := make(chan struct{}, readConcurrency) for i, ref := range refs { index := i stored := ref what := "reading the release " + ref.namespace + "/" + ref.name group.Add(1) safe.Go(what, func() { defer group.Done() defer func() { caught := recover() if caught == nil { return } safe.Log(what, caught) mu.Lock() out[index] = fallbackOf(stored) readable[index] = false mu.Unlock() }() slots <- struct{}{} defer func() { <-slots }() release, ok := s.releaseFor(ctx, stored) mu.Lock() out[index] = release readable[index] = ok mu.Unlock() }) } group.Wait() return out, readable } func (s *Service) releaseFor(ctx context.Context, ref storedRef) (api.HelmRelease, bool) { held, ok := s.cache.get(ref) if ok { return held, true } body, err := s.body(ctx, ref) if err != nil { return fallbackOf(ref), false } release, decodeErr := stored{ driver: ref.driver, namespace: ref.namespace, name: ref.name, revision: ref.revision, status: ref.status, created: ref.created, body: body, }.release() if decodeErr != nil { return release, false } s.cache.put(ref, release) return release, true } func fallbackOf(ref storedRef) api.HelmRelease { return stored{ namespace: ref.namespace, name: ref.name, revision: ref.revision, status: ref.status, created: ref.created, }.fallback() } func (s *Service) body(ctx context.Context, ref storedRef) ([]byte, error) { if ref.driver == DriverConfigMap { return configMapBody(ctx, s.cs, ref) } return secretBody(ctx, s.cs, ref) } func secretBody(ctx context.Context, cs kubernetes.Interface, ref storedRef) ([]byte, error) { got, err := cs.CoreV1().Secrets(ref.namespace).Get(ctx, ref.object, metav1.GetOptions{}) if err != nil { return nil, err } if got.Type != storageType { return nil, errNotRelease } return got.Data[releaseKey], nil } func configMapBody(ctx context.Context, cs kubernetes.Interface, ref storedRef) ([]byte, error) { got, err := cs.CoreV1().ConfigMaps(ref.namespace).Get(ctx, ref.object, metav1.GetOptions{}) if err != nil { return nil, err } body, ok := got.Data[releaseKey] if !ok { return nil, errNotRelease } return []byte(body), nil }