model/instance.go

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

package model

import (
	"github.com/coroot/coroot/timeseries"
)

type ClusterRole int

const (
	ClusterRoleNone ClusterRole = iota
	ClusterRolePrimary
	ClusterRoleReplica
	ClusterRoleArbiter
	ClusterRolePredictor
	ClusterRoleTransformer
	ClusterRoleExplainer
	ClusterRoleFrontend
	ClusterRoleWorker
	ClusterRolePrefill
	ClusterRoleDecode
	ClusterRolePlanner
	ClusterRoleEPP
)

var clusterRoleNames = map[ClusterRole]string{
	ClusterRolePrimary:     "primary",
	ClusterRoleReplica:     "replica",
	ClusterRoleArbiter:     "arbiter",
	ClusterRolePredictor:   "predictor",
	ClusterRoleTransformer: "transformer",
	ClusterRoleExplainer:   "explainer",
	ClusterRoleFrontend:    "frontend",
	ClusterRoleWorker:      "worker",
	ClusterRolePrefill:     "prefill",
	ClusterRoleDecode:      "decode",
	ClusterRolePlanner:     "planner",
	ClusterRoleEPP:         "epp",
}

var clusterRolesByName = func() map[string]ClusterRole {
	res := make(map[string]ClusterRole, len(clusterRoleNames))
	for r, name := range clusterRoleNames {
		res[name] = r
	}
	return res
}()

func (r ClusterRole) String() string {
	return clusterRoleNames[r]
}

type Instance struct {
	Name  string
	Owner *Application

	Node *Node

	Pod *Pod

	Rds         *Rds
	CloudSQL    *CloudSQL
	Memorystore *Memorystore
	OCIDB       *OCIDB
	OCICache    *OCICache
	Elasticache *Elasticache

	Jvms   map[string]*Jvm
	DotNet map[string]*DotNet
	Python *Python
	Nodejs *Nodejs
	Go     *GoRuntime

	Volumes []*Volume

	GPUUsage map[string]*InstanceGPUUsage

	Upstreams map[ConnectionKey]*Connection

	Requests Requests

	TcpListens map[Listen]bool

	Containers map[string]*Container

	Annotations ApplicationAnnotations

	ClusterName      LabelLastValue
	clusterRole      *timeseries.TimeSeries
	ClusterComponent *Application

	Postgres  *Postgres
	Redis     *Redis
	Mongodb   *Mongodb
	Memcached *Memcached
	Mysql     *Mysql
}

func NewInstance(name string, owner *Application) *Instance {
	return &Instance{
		Name:        name,
		Owner:       owner,
		Annotations: ApplicationAnnotations{},
		Containers:  map[string]*Container{},
		Upstreams:   map[ConnectionKey]*Connection{},
		TcpListens:  map[Listen]bool{},
		GPUUsage:    map[string]*InstanceGPUUsage{},
	}
}

func (instance *Instance) ApplicationTypes() map[ApplicationType]bool {
	res := map[ApplicationType]bool{}
	for _, c := range instance.Containers {
		for t := range c.ApplicationTypes {
			res[t] = true
		}
	}
	if t := instance.Rds.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	if t := instance.Elasticache.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	if t := instance.CloudSQL.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	if t := instance.Memorystore.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	if t := instance.OCIDB.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	if t := instance.OCICache.ApplicationType(); t != ApplicationTypeUnknown {
		res[t] = true
	}
	return res
}

func (instance *Instance) IsManagedDatabase() bool {
	return instance.Rds != nil || instance.Elasticache != nil || instance.CloudSQL != nil || instance.Memorystore != nil || instance.OCIDB != nil || instance.OCICache != nil
}

func (instance *Instance) InstrumentedType() ApplicationType {
	switch {
	case instance.Postgres != nil:
		return ApplicationTypePostgres
	case instance.Mysql != nil:
		return ApplicationTypeMysql
	case instance.Redis != nil:
		return ApplicationTypeRedis
	case instance.Mongodb != nil:
		return ApplicationTypeMongodb
	case instance.Memcached != nil:
		return ApplicationTypeMemcached
	}
	return ApplicationTypeUnknown
}

func (instance *Instance) GetOrCreateContainer(id, name string) *Container {
	c := instance.Containers[name]
	if c == nil {
		c = NewContainer(id, name)
		instance.Containers[name] = c
	}
	return c
}

func (instance *Instance) NodeName() string {
	if instance.Node != nil {
		return instance.Node.GetName()
	}
	return ""
}

func (instance *Instance) NodeId() NodeId {
	if instance.Node != nil {
		return instance.Node.Id
	}
	return NodeId{}
}

func (instance *Instance) UpdateClusterRole(role string, v *timeseries.TimeSeries) {
	r, ok := clusterRolesByName[role]
	if !ok {
		return
	}
	v = v.Map(func(t timeseries.Time, v float32) float32 {
		if v == 1 {
			return float32(r)
		}
		return timeseries.NaN
	})
	if instance.clusterRole == nil {
		instance.clusterRole = v
	} else {
		instance.clusterRole = timeseries.NewAggregate(timeseries.Any).Add(instance.clusterRole, v).Get()
	}
}

func (instance *Instance) ClusterRole() *timeseries.TimeSeries {
	if instance.Pod == nil || instance.Pod.Ready.IsEmpty() || instance.clusterRole.IsEmpty() {
		return instance.clusterRole
	}
	return timeseries.Mul(instance.clusterRole, instance.Pod.Ready)
}

func (instance *Instance) ClusterRoleLast() ClusterRole {
	role := instance.ClusterRole()
	if role.IsEmpty() {
		return ClusterRoleNone
	}
	return ClusterRole(role.Last())
}

func (instance *Instance) LifeSpan() *timeseries.TimeSeries {
	if instance.Pod != nil {
		return instance.Pod.LifeSpan
	}
	for _, c := range instance.Containers {
		return c.MemoryRss.Map(timeseries.Defined)
	}
	return nil
}

func (instance *Instance) IsUp() bool {
	for _, c := range instance.Containers {
		if !c.MemoryRss.TailIsEmpty() {
			return true
		}
	}
	return false
}

func (instance *Instance) IsObsolete() bool {
	return instance.Pod != nil && instance.Pod.IsObsolete()
}

func (instance *Instance) IsFailed() bool {
	return instance.Pod != nil && instance.Pod.IsFailed()
}

func (instance *Instance) IsJobPod() bool {
	return instance.Pod != nil &&
		(instance.Pod.OwnerKind == ApplicationKindJob || instance.Pod.OwnerKind == ApplicationKindCronJob)
}

func (instance *Instance) UpAndRunning() *timeseries.TimeSeries {
	mem := timeseries.NewAggregate(timeseries.Any)
	for _, c := range instance.Containers {
		mem.Add(c.MemoryRss)
	}
	memTs := mem.Get()

	if memTs.IsEmpty() {
		return nil
	}
	up := memTs.Map(func(t timeseries.Time, v float32) float32 {
		if v > 0 {
			return 1
		}
		return 0
	})
	if instance.Pod == nil {
		return up
	}
	running := instance.Pod.Running.Map(timeseries.NanToZero)
	ready := instance.Pod.Ready.Map(timeseries.NanToZero)
	return timeseries.NewAggregate(timeseries.Min).Add(running, ready, up).Get()
}

func (instance *Instance) IsListenActive(ip, port string) bool {
	for l, active := range instance.TcpListens {
		if l.IP == ip && (l.Port == port || l.Port == "0") {
			return active
		}
	}
	return false
}

func (instance *Instance) NodeContainerID() *NodeContainerId {
	if instance.Node == nil {
		return nil
	}
	for _, c := range instance.Containers {
		return &NodeContainerId{
			NodeId:      instance.Node.Id,
			ContainerId: c.Id,
		}
	}
	return nil
}

type Listen struct {
	IP      string
	Port    string
	Proxied bool
}

type Requests struct {
	Ok           *timeseries.TimeSeries
	Failed       *timeseries.TimeSeries
	TotalLatency *timeseries.TimeSeries
}