internal/resources/timeline.go

Rama main · réplica de lectura del repositorio original.

package resources

import (
	"context"
	"hash/fnv"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/sophotechlabs/spinoza/internal/api"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
)

const (
	Added   = "added"
	Changed = "changed"
	Removed = "removed"
)

type Note struct {
	At        time.Time
	Verb      string
	Group     string
	Version   string
	Resource  string
	Kind      string
	Namespace string
	Name      string
	UID       string
	Cells     []string
	Was       []string
}

type Timeline interface {
	Note(note Note)
}

type Kind struct {
	Group    string
	Resource string
}

func (m *Manager) Record(ctx context.Context, into Timeline, kinds []Kind) {
	descs := m.recorded(kinds)
	wanted := map[schema.GroupVersionResource]struct{}{}
	for _, desc := range descs {
		wanted[gvrOf(desc)] = struct{}{}
	}
	m.noteMu.Lock()
	m.notes = into
	m.noted = wanted
	m.noteMu.Unlock()
	m.Warm(ctx, descs)
}

func (m *Manager) recorded(kinds []Kind) []api.ResourceDescriptor {
	known := m.descriptors()
	out := make([]api.ResourceDescriptor, 0, len(kinds))
	for _, want := range kinds {
		for _, desc := range known {
			if desc.Group == want.Group && desc.Resource == want.Resource {
				out = append(out, desc)
				break
			}
		}
	}
	slices.SortFunc(out, func(left, right api.ResourceDescriptor) int {
		return strings.Compare(gvrOf(left).String(), gvrOf(right).String())
	})
	return out
}

func (m *Manager) StopRecording() {
	m.noteMu.Lock()
	was := m.noted
	m.notes = nil
	m.noted = nil
	m.noteMu.Unlock()
	for gvr := range was {
		m.unpin(streamKey{gvr: gvr})
	}
}

func (m *Manager) record(gvr schema.GroupVersionResource, note Note) {
	m.noteMu.Lock()
	defer m.noteMu.Unlock()
	if m.notes == nil {
		return
	}
	_, wanted := m.noted[gvr]
	if !wanted {
		return
	}
	m.notes.Note(note)
}

func gvrOf(desc api.ResourceDescriptor) schema.GroupVersionResource {
	return schema.GroupVersionResource{
		Group:    desc.Group,
		Version:  desc.Version,
		Resource: desc.Resource,
	}
}

func (st *stream) note(verb string, obj *unstructured.Unstructured, row api.Row) {
	was, fresh := st.rowIsNew(verb, row)
	if !fresh {
		return
	}
	if !st.delivered() {
		return
	}
	st.owner.record(st.gvr, Note{
		At:        time.Now().UTC(),
		Verb:      verb,
		Group:     st.gvr.Group,
		Version:   st.gvr.Version,
		Resource:  st.gvr.Resource,
		Kind:      st.kind,
		Namespace: obj.GetNamespace(),
		Name:      obj.GetName(),
		UID:       string(obj.GetUID()),
		Cells:     row.Cells,
		Was:       was,
	})
}

func (st *stream) delivered() bool {
	handler := st.handler.Load()
	if handler == nil {
		return false
	}
	return (*handler).HasSynced()
}

func (st *stream) rowIsNew(verb string, row api.Row) ([]string, bool) {
	st.seenMu.Lock()
	defer st.seenMu.Unlock()
	if verb == Removed {
		last, known := st.seen[row.UID]
		delete(st.seen, row.UID)
		if !known {
			return nil, true
		}
		return last.cells, true
	}
	if st.seen == nil {
		st.seen = map[string]held{}
	}
	shape := shapeOf(row)
	was, known := st.seen[row.UID]
	st.seen[row.UID] = held{shape: shape, cells: row.Cells}
	if !known {
		return nil, true
	}
	if was.shape == shape {
		return nil, false
	}
	return was.cells, true
}

type held struct {
	shape uint64
	cells []string
}

func shapeOf(row api.Row) uint64 {
	digest := fnv.New64a()
	for _, cell := range row.Cells {
		_, _ = digest.Write([]byte(cell))
		_, _ = digest.Write([]byte{0})
	}
	for _, one := range row.Containers {
		_, _ = digest.Write([]byte(one.Name + "\x00" + one.State + "\x00" + one.Reason))
		_, _ = digest.Write([]byte(strconv.FormatBool(one.Ready)))
		_, _ = digest.Write([]byte(strconv.FormatInt(one.Restarts, 10)))
		_, _ = digest.Write([]byte{0})
	}
	return digest.Sum64()
}