240 lines
7.0 KiB
Go
240 lines
7.0 KiB
Go
// Package kubernetes observes read-only workload state for rollout correlation.
|
|
package kubernetes
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"os"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
appsv1 "k8s.io/api/apps/v1"
|
|
corev1 "k8s.io/api/core/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/client-go/informers"
|
|
"k8s.io/client-go/kubernetes"
|
|
"k8s.io/client-go/rest"
|
|
"k8s.io/client-go/tools/cache"
|
|
"k8s.io/client-go/tools/clientcmd"
|
|
)
|
|
|
|
const maxRollouts = 500
|
|
|
|
type Watcher struct {
|
|
mu sync.RWMutex
|
|
rollouts map[string]Rollout
|
|
enabled bool
|
|
err string
|
|
}
|
|
|
|
// NewFromEnv enables observation in-cluster, or with KUBECONFIG locally. It is
|
|
// disabled by default outside Kubernetes so local development needs no kubeconfig.
|
|
func NewFromEnv(ctx context.Context) (*Watcher, error) {
|
|
if strings.EqualFold(os.Getenv("OTEL_KUBERNETES_ENABLED"), "false") {
|
|
return nil, nil
|
|
}
|
|
config, err := kubeConfig()
|
|
if err != nil {
|
|
if os.Getenv("OTEL_KUBERNETES_ENABLED") == "true" {
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}
|
|
client, err := kubernetes.NewForConfig(config)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create kubernetes client: %w", err)
|
|
}
|
|
|
|
w := &Watcher{rollouts: make(map[string]Rollout), enabled: true}
|
|
namespace := os.Getenv("OTEL_KUBERNETES_NAMESPACE")
|
|
selector := os.Getenv("OTEL_KUBERNETES_LABEL_SELECTOR")
|
|
factory := informers.NewSharedInformerFactoryWithOptions(client, 30*time.Second,
|
|
informers.WithTweakListOptions(func(options *metav1.ListOptions) {
|
|
options.LabelSelector = selector
|
|
}),
|
|
)
|
|
if namespace != "" {
|
|
factory = informers.NewSharedInformerFactoryWithOptions(client, 30*time.Second,
|
|
informers.WithNamespace(namespace), informers.WithTweakListOptions(func(options *metav1.ListOptions) {
|
|
options.LabelSelector = selector
|
|
}))
|
|
}
|
|
informer := factory.Apps().V1().Deployments().Informer()
|
|
_, _ = informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
|
|
AddFunc: func(obj interface{}) { w.observe(obj) },
|
|
UpdateFunc: func(_, obj interface{}) { w.observe(obj) },
|
|
})
|
|
go func() {
|
|
factory.Start(ctx.Done())
|
|
if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
|
|
w.setError("deployment informer did not sync")
|
|
return
|
|
}
|
|
<-ctx.Done()
|
|
}()
|
|
return w, nil
|
|
}
|
|
|
|
func kubeConfig() (*rest.Config, error) {
|
|
if config, err := rest.InClusterConfig(); err == nil {
|
|
return config, nil
|
|
}
|
|
kubeconfig := os.Getenv("KUBECONFIG")
|
|
if kubeconfig == "" {
|
|
home, err := os.UserHomeDir()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kubeconfig = home + string(os.PathSeparator) + ".kube" + string(os.PathSeparator) + "config"
|
|
}
|
|
return clientcmd.BuildConfigFromFlags("", kubeconfig)
|
|
}
|
|
|
|
func (w *Watcher) observe(obj interface{}) {
|
|
deployment, ok := obj.(*appsv1.Deployment)
|
|
if !ok {
|
|
return
|
|
}
|
|
now := time.Now().UnixNano()
|
|
revision := deployment.Annotations["deployment.kubernetes.io/revision"]
|
|
if revision == "" {
|
|
revision = strconv.FormatInt(deployment.Generation, 10)
|
|
}
|
|
key := deployment.Namespace + "/" + deployment.Name + "/" + revision
|
|
phase, reason, message, completed := rolloutStatus(deployment)
|
|
started := now
|
|
for _, condition := range deployment.Status.Conditions {
|
|
if condition.Type == appsv1.DeploymentProgressing && !condition.LastUpdateTime.IsZero() {
|
|
started = condition.LastUpdateTime.UnixNano()
|
|
}
|
|
if condition.Type == appsv1.DeploymentAvailable && !condition.LastUpdateTime.IsZero() && completed {
|
|
completed = condition.LastUpdateTime.UnixNano() != 0
|
|
}
|
|
}
|
|
if old, exists := w.rollouts[key]; exists {
|
|
started = old.StartedAt
|
|
if old.CompletedAt != 0 {
|
|
completed = true
|
|
}
|
|
}
|
|
rollout := Rollout{
|
|
ID: rolloutID(key), Namespace: deployment.Namespace, Workload: deployment.Name,
|
|
Service: metadataValue(deployment, "app.kubernetes.io/name", "app", "service.name"),
|
|
Release: metadataValue(deployment, "otel.k6n.net/release", "app.kubernetes.io/version", "service.release"),
|
|
Revision: revision, Image: deploymentImage(deployment), Phase: phase, Reason: reason,
|
|
Message: message, StartedAt: started, LastSeen: now, Generation: deployment.Generation,
|
|
DesiredReplicas: desiredReplicas(deployment), UpdatedReplicas: deployment.Status.UpdatedReplicas,
|
|
AvailableReplicas: deployment.Status.AvailableReplicas, ReadyReplicas: deployment.Status.ReadyReplicas,
|
|
}
|
|
if completed {
|
|
rollout.CompletedAt = now
|
|
}
|
|
w.mu.Lock()
|
|
w.rollouts[key] = rollout
|
|
if len(w.rollouts) > maxRollouts {
|
|
w.evictOldestLocked()
|
|
}
|
|
w.mu.Unlock()
|
|
}
|
|
|
|
func (w *Watcher) evictOldestLocked() {
|
|
var oldest string
|
|
var ts int64
|
|
for key, item := range w.rollouts {
|
|
if oldest == "" || item.LastSeen < ts {
|
|
oldest, ts = key, item.LastSeen
|
|
}
|
|
}
|
|
if oldest != "" {
|
|
delete(w.rollouts, oldest)
|
|
}
|
|
}
|
|
func (w *Watcher) setError(message string) { w.mu.Lock(); defer w.mu.Unlock(); w.err = message }
|
|
|
|
func (w *Watcher) List() []Rollout {
|
|
w.mu.RLock()
|
|
defer w.mu.RUnlock()
|
|
result := make([]Rollout, 0, len(w.rollouts))
|
|
for _, item := range w.rollouts {
|
|
result = append(result, item)
|
|
}
|
|
sort.Slice(result, func(i, j int) bool { return result[i].LastSeen > result[j].LastSeen })
|
|
return result
|
|
}
|
|
func (w *Watcher) Get(id string) (Rollout, bool) {
|
|
w.mu.RLock()
|
|
defer w.mu.RUnlock()
|
|
for _, item := range w.rollouts {
|
|
if item.ID == id {
|
|
return item, true
|
|
}
|
|
}
|
|
return Rollout{}, false
|
|
}
|
|
func (w *Watcher) Enabled() bool { return w != nil && w.enabled }
|
|
func (w *Watcher) Error() string {
|
|
if w == nil {
|
|
return "kubernetes integration is not configured"
|
|
}
|
|
w.mu.RLock()
|
|
defer w.mu.RUnlock()
|
|
return w.err
|
|
}
|
|
|
|
func rolloutID(key string) string {
|
|
sum := sha256.Sum256([]byte(key))
|
|
return hex.EncodeToString(sum[:8])
|
|
}
|
|
func metadataValue(deployment *appsv1.Deployment, keys ...string) string {
|
|
for _, key := range keys {
|
|
if value := deployment.Labels[key]; value != "" {
|
|
return value
|
|
}
|
|
if value := deployment.Annotations[key]; value != "" {
|
|
return value
|
|
}
|
|
if value := deployment.Spec.Template.Labels[key]; value != "" {
|
|
return value
|
|
}
|
|
if value := deployment.Spec.Template.Annotations[key]; value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
func deploymentImage(deployment *appsv1.Deployment) string {
|
|
if len(deployment.Spec.Template.Spec.Containers) == 0 {
|
|
return ""
|
|
}
|
|
return deployment.Spec.Template.Spec.Containers[0].Image
|
|
}
|
|
func desiredReplicas(deployment *appsv1.Deployment) int32 {
|
|
if deployment.Spec.Replicas == nil {
|
|
return 1
|
|
}
|
|
return *deployment.Spec.Replicas
|
|
}
|
|
|
|
func rolloutStatus(deployment *appsv1.Deployment) (phase, reason, message string, completed bool) {
|
|
phase = "progressing"
|
|
for _, condition := range deployment.Status.Conditions {
|
|
switch condition.Type {
|
|
case appsv1.DeploymentProgressing:
|
|
reason, message = condition.Reason, condition.Message
|
|
if condition.Status == corev1.ConditionFalse {
|
|
return "failed", reason, message, false
|
|
}
|
|
case appsv1.DeploymentAvailable:
|
|
if condition.Status == corev1.ConditionTrue && deployment.Status.UpdatedReplicas == desiredReplicas(deployment) && deployment.Status.AvailableReplicas >= desiredReplicas(deployment) {
|
|
return "completed", reason, message, true
|
|
}
|
|
}
|
|
}
|
|
return phase, reason, message, false
|
|
}
|