constructor/constructor.go

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

package constructor

import (
	"context"
	"fmt"
	"net"
	"strings"
	"sync"
	"time"

	pricing "github.com/coroot/coroot/cloud-pricing"
	"github.com/coroot/coroot/db"
	"github.com/coroot/coroot/model"
	"github.com/coroot/coroot/timeseries"
	"github.com/coroot/coroot/utils"
	"k8s.io/klog"
)

type Option int

const (
	OptionLoadInstanceToInstanceConnections Option = iota
	OptionDoNotLoadRawSLIs
	OptionLoadContainerLogs
	OptionDoNotLoadAppDBData
)

type DB interface {
	GetCheckConfigs(projectId db.ProjectId) (model.CheckConfigs, error)
	GetApplicationDeployments(projectId db.ProjectId) (map[model.ApplicationId][]*model.ApplicationDeployment, error)
	GetApplicationIncidents(projectId db.ProjectId, from, to timeseries.Time) (map[model.ApplicationId][]*model.ApplicationIncident, error)
	GetApplicationSettingsByProject(projectId db.ProjectId) (map[model.ApplicationId]*model.ApplicationSettings, error)
	GetProjects() (map[string]*db.Project, error)
}

type Cache interface {
	QueryRange(ctx context.Context, query string, from, to timeseries.Time, step timeseries.Duration, fillFunc timeseries.FillFunc) ([]*model.MetricValues, error)
	GetStep(from, to timeseries.Time) (timeseries.Duration, error)
}

type Constructor struct {
	db           DB
	rootProject  *db.Project
	cacheClients map[db.ProjectId]Cache
	pricing      *pricing.Manager
	options      map[Option]bool
}

func New(db DB, project *db.Project, cacheClients map[db.ProjectId]Cache, pricing *pricing.Manager, options ...Option) *Constructor {
	c := &Constructor{db: db, rootProject: project, cacheClients: cacheClients, pricing: pricing, options: map[Option]bool{}}
	for _, o := range options {
		c.options[o] = true
	}
	return c
}

func (c *Constructor) newApplicationId(clusterId string, namespace string, kind model.ApplicationKind, name string) model.ApplicationId {
	return model.NewApplicationId(clusterId, namespace, kind, name)
}

func (c *Constructor) LoadWorld(ctx context.Context, from, to timeseries.Time, step timeseries.Duration, prof *Profile) (*model.World, error) {
	if !c.rootProject.Multicluster() {
		cache, ok := c.cacheClients[c.rootProject.Id]
		if !ok {
			return nil, fmt.Errorf("metrics cache is missing for project %s", c.rootProject.Id)
		}
		return c.loadProjectWorld(ctx, cache, c.rootProject, from, to, step, prof, nil)
	}
	projects, err := c.db.GetProjects()
	if err != nil {
		return nil, err
	}
	var memberProjects []*db.Project
	for _, pName := range c.rootProject.Settings.MemberProjects {
		p := projects[pName]
		if p == nil {
			return nil, fmt.Errorf("project %s not found", pName)
		}
		memberProjects = append(memberProjects, p)
	}
	var worlds []*model.World
	for _, mp := range memberProjects {
		cache, ok := c.cacheClients[mp.Id]
		if !ok {
			return nil, fmt.Errorf("metrics cache is missing for project %s", mp.Id)
		}
		w, err := c.loadProjectWorld(ctx, cache, mp, from, to, step, prof, c.rootProject)
		if err != nil {
			return nil, err
		}
		worlds = append(worlds, w)
	}
	return mergeWorlds(worlds), nil
}

func (c *Constructor) loadProjectWorld(ctx context.Context, cache Cache, project *db.Project, from, to timeseries.Time, step timeseries.Duration, prof *Profile, parentProject *db.Project) (*model.World, error) {
	start := time.Now()
	rawStep, err := cache.GetStep(from, to)
	if err != nil {
		return nil, err
	}
	if rawStep == 0 {
		return model.NewWorld(from, to, step, step), nil
	}

	w := model.NewWorld(from, to, step, rawStep)
	w.ProjectNamesById[string(project.Id)] = project.Name

	if prof == nil {
		prof = &Profile{}
	}

	prof.stage("get_check_configs", func() {
		w.CheckConfigs, err = c.db.GetCheckConfigs(project.Id)
	})
	if err != nil {
		return nil, err
	}

	var metrics map[string][]*model.MetricValues
	prof.stage("query", func() {
		metrics, err = c.queryCache(ctx, cache, project, from, to, step, rawStep, w.CheckConfigs, prof.Queries)
	})
	if err != nil {
		return nil, err
	}

	pjs := promJobStatuses{}
	nodes := nodeCache{}
	cloudInstancesById := map[string]*model.Instance{}
	servicesByClusterIP := map[string]*model.Service{}
	ip2fqdn := map[string]*utils.StringSet{}
	fqdn2ip := map[string]*utils.StringSet{}
	containers := containerCache{}
	inferenceGroups := map[model.ApplicationId]model.ApplicationId{}

	// order is important
	prof.stage("load_job_statuses", func() { loadPromJobStatuses(metrics, pjs) })
	prof.stage("load_nodes", func() { c.loadNodes(w, metrics, nodes, project) })
	prof.stage("load_fqdn", func() { loadFQDNs(metrics, ip2fqdn, fqdn2ip) })
	prof.stage("load_fargate_nodes", func() { c.loadFargateNodes(metrics, nodes) })
	prof.stage("load_k8s_metadata", func() { c.loadKubernetesMetadata(w, metrics, servicesByClusterIP, inferenceGroups, project) })
	prof.stage("load_flux_resources", func() { loadFluxResources(w, metrics, project) })
	prof.stage("load_argocd_resources", func() { loadArgoCDResources(w, metrics, project) })
	prof.stage("load_cloud_status", func() { loadCloudStatus(w, metrics) })
	prof.stage("load_rds_metadata", func() { c.loadRdsMetadata(w, metrics, pjs, cloudInstancesById, project) })
	prof.stage("load_elasticache_metadata", func() { c.loadElasticacheMetadata(w, metrics, pjs, cloudInstancesById, project) })
	prof.stage("load_rds", func() { c.loadRds(w, metrics, pjs, cloudInstancesById) })
	prof.stage("load_elasticache", func() { c.loadElasticache(w, metrics, pjs, cloudInstancesById) })
	prof.stage("load_gcp_metadata", func() { c.loadGCPMetadata(w, metrics, cloudInstancesById, project) })
	prof.stage("load_gcp", func() { c.loadGCP(w, metrics, pjs, cloudInstancesById) })
	prof.stage("load_oci_metadata", func() { c.loadOCIMetadata(w, metrics, cloudInstancesById, project) })
	prof.stage("load_oci", func() { c.loadOCI(w, metrics, pjs, cloudInstancesById) })
	prof.stage("load_fargate_containers", func() { loadFargateContainers(w, metrics, pjs) })
	prof.stage("load_containers", func() { c.loadContainers(w, metrics, pjs, nodes, containers, servicesByClusterIP, ip2fqdn, project) })
	prof.stage("load_app_to_app_connections", func() { c.loadAppToAppConnections(w, metrics, fqdn2ip, project) })
	prof.stage("load_application_traffic", func() { c.loadApplicationTraffic(w, metrics, project) })
	prof.stage("load_application_dns", func() { c.loadApplicationDNS(w, metrics, project) })
	prof.stage("load_jvm", func() { c.loadJVM(metrics, containers) })
	prof.stage("load_go_runtime", func() { c.loadGoRuntime(metrics, containers) })
	prof.stage("load_dotnet", func() { c.loadDotNet(metrics, containers) })
	prof.stage("load_python", func() { c.loadPython(metrics, containers) })
	prof.stage("load_nodejs", func() { c.loadNodejs(metrics, containers) })
	var instancesByListen map[model.Listen]*model.Instance
	prof.stage("enrich_instances", func() {
		instancesByListen = enrichInstances(w, metrics, cloudInstancesById, pjs, containers)
	})
	prof.stage("calc_app_categories", func() { c.calcApplicationCategories(w, project) })
	prof.stage("group_custom_applications", func() { c.groupCustomApplications(w, project) })
	prof.stage("group_inference_applications", func() { c.groupInferenceApplications(w, inferenceGroups, project) })
	prof.stage("join_db_cluster_components", func() { c.joinDBClusterComponents(w, project) })
	prof.stage("merge_external_into_pods", func() { mergeExternalServicesIntoPods(w, instancesByListen) })
	prof.stage("load_postgres_backups", func() { loadPostgresBackups(w, metrics, project) })
	prof.stage("load_mongo_backups", func() { loadMongoBackups(w, metrics, project) })
	prof.stage("load_mysql_backups", func() { loadMysqlBackups(w, metrics) })
	prof.stage("load_app_sli", func() { c.loadSLIs(w, metrics, project) })
	prof.stage("load_container_logs", func() { c.loadContainerLogs(metrics, containers, pjs) })
	prof.stage("load_app_logs", func() { c.loadApplicationLogs(w, metrics, project) })
	if !c.options[OptionDoNotLoadAppDBData] {
		prof.stage("load_app_settings", func() { c.loadApplicationSettings(w, project) })
		prof.stage("load_app_deployments", func() { c.loadApplicationDeployments(w, project) })
		prof.stage("load_app_incidents", func() { c.loadApplicationIncidents(w, project, parentProject) })
	}
	prof.stage("calc_app_events", func() { calcAppEvents(w) })

	klog.Infof("%s: got %d nodes, %d apps in %s", project.Id, len(w.Nodes), len(w.Applications), time.Since(start).Truncate(time.Millisecond))
	return w, nil
}

func (c *Constructor) QueryCache(ctx context.Context, cache Cache, project *db.Project, from, to timeseries.Time, step timeseries.Duration) (map[string][]*model.MetricValues, error) {
	rawStep, err := cache.GetStep(from, to)
	if err != nil {
		return nil, err
	}
	if rawStep == 0 {
		return nil, nil
	}
	checkConfigs, err := c.db.GetCheckConfigs(project.Id)
	if err != nil {
		return nil, err
	}
	return c.queryCache(ctx, cache, project, from, to, step, rawStep, checkConfigs, nil)
}

type cacheQuery struct {
	query     string
	from, to  timeseries.Time
	step      timeseries.Duration
	statsName string
	fillFunc  timeseries.FillFunc
}

func (c *Constructor) queryCache(ctx context.Context, cache Cache, project *db.Project, from, to timeseries.Time, step, rawStep timeseries.Duration, checkConfigs model.CheckConfigs, stats map[string]QueryStats) (map[string][]*model.MetricValues, error) {
	loadRawSLIs := !c.options[OptionDoNotLoadRawSLIs]
	rawFrom := to.Add(-model.MaxAlertRuleWindow).Truncate(rawStep)
	rawTo := to.Truncate(rawStep)

	from = from.Truncate(step)
	to = to.Truncate(step)
	queries := map[string]cacheQuery{}
	addQuery := func(name, statsName, query string, sli bool, fillFunc timeseries.FillFunc) {
		if sli && loadRawSLIs {
			queries[name+"_raw"] = cacheQuery{query: query, from: rawFrom, to: rawTo, step: rawStep, statsName: statsName + "_raw"}
			queries[name+"_avg"] = cacheQuery{query: query, from: from, to: to, step: step, statsName: statsName + "_avg", fillFunc: timeseries.FillAvg}
		}
		if !sli {
			queries[name] = cacheQuery{query: query, from: from, to: to, step: step, statsName: statsName, fillFunc: fillFunc}
		}
	}

	for _, q := range QUERIES {
		if !c.options[OptionLoadInstanceToInstanceConnections] && q.InstanceToInstance {
			continue
		}
		if !c.options[OptionLoadContainerLogs] && q.Name == "container_log_messages" {
			queries[qRecordingRuleApplicationLogMessages] = cacheQuery{
				query:     qRecordingRuleApplicationLogMessages,
				from:      from,
				to:        to,
				step:      step,
				statsName: qRecordingRuleApplicationLogMessages,
				fillFunc:  timeseries.FillSum,
			}
			continue
		}
		addQuery(q.Name, q.Name, q.Query, false, q.FillFunc)
	}
	if !c.options[OptionLoadInstanceToInstanceConnections] {
		for _, query := range qConnectionAggregations {
			queries[query] = cacheQuery{query: query, from: from, to: to, step: step, statsName: query}
		}
		addQuery(qRecordingRuleApplicationL7Requests, qRecordingRuleApplicationL7Requests, qRecordingRuleApplicationL7Requests, true, nil)
		addQuery(qRecordingRuleApplicationL7Histogram, qRecordingRuleApplicationL7Histogram, qRecordingRuleApplicationL7Histogram, true, nil)
		addQuery(qRecordingRuleApplicationL7InboundRequests, qRecordingRuleApplicationL7InboundRequests, qRecordingRuleApplicationL7InboundRequests, true, nil)
		addQuery(qRecordingRuleApplicationL7InboundHistogram, qRecordingRuleApplicationL7InboundHistogram, qRecordingRuleApplicationL7InboundHistogram, true, nil)
		addQuery(qRecordingRuleApplicationLLMInboundHistogram, qRecordingRuleApplicationLLMInboundHistogram, qRecordingRuleApplicationLLMInboundHistogram, true, nil)
	}
	for appId := range checkConfigs {
		qName := fmt.Sprintf("%s/%s/", qApplicationCustomSLI, appId)
		availabilityCfg, _ := checkConfigs.GetAvailability(appId)
		if availabilityCfg.Custom {
			addQuery(qName+"total_requests", qApplicationCustomSLI, availabilityCfg.Total(), true, nil)
			addQuery(qName+"failed_requests", qApplicationCustomSLI, availabilityCfg.Failed(), true, nil)
		}
		latencyCfg, _ := checkConfigs.GetLatency(appId, project.CalcApplicationCategory(appId))
		if latencyCfg.Custom {
			addQuery(qName+"requests_histogram", qApplicationCustomSLI, latencyCfg.Histogram(), true, nil)
		}
	}

	res := make(map[string][]*model.MetricValues, len(queries))
	var lock sync.Mutex
	var lastErr error
	wg := sync.WaitGroup{}
	now := time.Now()
	for name, query := range queries {
		wg.Add(1)
		go func(name string, q cacheQuery) {
			defer wg.Done()
			if q.fillFunc == nil {
				q.fillFunc = timeseries.FillAny
			}
			metrics, err := cache.QueryRange(ctx, q.query, q.from, q.to, q.step, q.fillFunc)
			if stats != nil {
				queryTime := float32(time.Since(now).Seconds())
				lock.Lock()
				s := stats[q.statsName]
				s.MetricsCount += len(metrics)
				s.QueryTime += queryTime
				s.Failed = s.Failed || err != nil
				s.Cardinality = cardinalityStats(metrics)
				stats[q.statsName] = s
				lock.Unlock()
			}
			if err != nil {
				lastErr = err
				return
			}
			lock.Lock()
			res[name] = metrics
			lock.Unlock()
		}(name, query)
	}
	wg.Wait()
	return res, lastErr
}

func (c *Constructor) calcApplicationCategories(w *model.World, project *db.Project) {
	for _, app := range w.Applications {
		if annotation := app.GetAnnotation(model.ApplicationAnnotationCategory); annotation != "" {
			app.Category = model.ApplicationCategory(annotation)
			continue
		}
		app.Category = project.CalcApplicationCategory(app.Id)
	}
}

func (c *Constructor) loadApplicationSettings(w *model.World, project *db.Project) {
	settings, err := c.db.GetApplicationSettingsByProject(project.Id)
	if err != nil {
		klog.Errorln(err)
		return
	}
	for _, app := range w.Applications {
		app.Settings = settings[app.Id]
	}
}

func (c *Constructor) loadApplicationDeployments(w *model.World, project *db.Project) {
	byApp, err := c.db.GetApplicationDeployments(project.Id)
	if err != nil {
		klog.Errorln(err)
		return
	}
	for id, deployments := range byApp {
		app := w.GetApplication(id)
		if app == nil {
			continue
		}
		app.Deployments = deployments
	}
}

func (c *Constructor) loadApplicationIncidents(w *model.World, project, parentProject *db.Project) {
	projectId := project.Id
	if parentProject != nil {
		projectId = parentProject.Id
	}
	byApp, err := c.db.GetApplicationIncidents(projectId, w.Ctx.From, w.Ctx.To)
	if err != nil {
		klog.Errorln(err)
		return
	}
	for id, incidents := range byApp {
		app := w.GetApplication(id)
		if app == nil {
			continue
		}
		app.Incidents = incidents
	}
}

type promJob struct {
	job      string
	instance string
}

type promJobStatuses map[promJob]*timeseries.TimeSeries

func (s promJobStatuses) get(ls model.Labels) *timeseries.TimeSeries {
	return s[promJob{job: ls["job"], instance: ls["instance"]}]
}

func loadPromJobStatuses(metrics map[string][]*model.MetricValues, statuses promJobStatuses) {
	for _, m := range metrics["up"] {
		statuses[promJob{job: m.Labels["job"], instance: m.Labels["instance"]}] = m.Values
	}
}

type podId struct {
	name, ns string
}

func enrichInstances(w *model.World, metrics map[string][]*model.MetricValues, cloudInstancesById map[string]*model.Instance, pjs promJobStatuses, containers containerCache) map[model.Listen]*model.Instance {
	instancesByListen := map[model.Listen]*model.Instance{}
	instancesByPod := map[podId]*model.Instance{}
	for _, app := range w.Applications {
		for _, i := range app.Instances {
			if i.Pod != nil {
				instancesByPod[podId{name: i.Name, ns: app.Id.Namespace}] = i
			}
			for l := range i.TcpListens {
				if e := instancesByListen[l]; e == nil || (e.Owner.Id.Kind == model.ApplicationKindExternalService && app.Id.Kind != model.ApplicationKindExternalService) {
					instancesByListen[l] = i
				}
			}
		}
	}

	betterInstance := func(cur, candidate *model.Instance) *model.Instance {
		switch {
		case cur == nil:
			return candidate
		case candidate == nil:
			return cur
		case cur.IsObsolete() != candidate.IsObsolete():
			if cur.IsObsolete() {
				return candidate
			}
			return cur
		case candidate.Name < cur.Name:
			return candidate
		}
		return cur
	}
	instancesByListenAddr := map[string]*model.Instance{}
	for l, i := range instancesByListen {
		if ip := net.ParseIP(l.IP); ip == nil {
			continue
		}
		addr := net.JoinHostPort(l.IP, l.Port)
		l.Proxied = true
		if ii := instancesByListen[l]; ii != nil {
			i = ii
		} else {
			l.Proxied = false
			if ii := instancesByListen[l]; ii != nil {
				i = ii
			}
		}
		instancesByListenAddr[addr] = betterInstance(instancesByListenAddr[addr], i)
	}

	for _, queryName := range []string{"pg_up", "redis_up", "mongo_up", "memcached_up"} {
		for _, m := range metrics[queryName] {
			address := m.Labels["address"]
			if address == "" {
				continue
			}
			instance := instancesByListenAddr[address]
			if instance == nil {
				continue
			}
			switch queryName {
			case "pg_up":
				instance.Postgres = model.NewPostgres()
			case "redis_up":
				instance.Redis = model.NewRedis()
			case "mongo_up":
				instance.Mongodb = model.NewMongodb()
			case "memcached_up":
				instance.Memcached = model.NewMemcached()
			}
		}
	}

	for queryName := range metrics {
		for _, m := range metrics[queryName] {
			switch {
			case strings.HasPrefix(queryName, "pg_"):
				instance := findInstance(instancesByPod, instancesByListenAddr, cloudInstancesById, containers, m, model.ApplicationTypePostgres)
				postgres(instance, queryName, m, pjs)
			case strings.HasPrefix(queryName, "redis_"):
				instance := findInstance(instancesByPod, instancesByListenAddr, cloudInstancesById, containers, m, model.ApplicationTypeRedis, model.ApplicationTypeKeyDB)
				redis(instance, queryName, m)
			case strings.HasPrefix(queryName, "mongo_"):
				instance := findInstance(instancesByPod, instancesByListenAddr, cloudInstancesById, containers, m, model.ApplicationTypeMongodb, model.ApplicationTypeMongos)
				mongodb(instance, queryName, m, pjs)
			case strings.HasPrefix(queryName, "memcached_"):
				instance := findInstance(instancesByPod, instancesByListenAddr, cloudInstancesById, containers, m, model.ApplicationTypeMemcached)
				memcached(instance, queryName, m)
			case strings.HasPrefix(queryName, "mysql_"):
				instance := findInstance(instancesByPod, instancesByListenAddr, cloudInstancesById, containers, m, model.ApplicationTypeMysql)
				mysql(instance, queryName, m)
			}
		}
	}
	return instancesByListen
}

func mergeExternalServicesIntoPods(w *model.World, instancesByListen map[model.Listen]*model.Instance) {
	for extId, ext := range w.Applications {
		if ext.Id.Kind != model.ApplicationKindExternalService {
			continue
		}
		var target *model.Instance
		match := len(ext.Instances) > 0
		for _, ei := range ext.Instances {
			for l := range ei.TcpListens {
				i := instancesByListen[model.Listen{IP: l.IP, Port: "0"}]
				if i == nil || (target != nil && target.Owner != i.Owner) {
					match = false
				} else {
					target = i
				}
			}
		}
		if !match || target == nil {
			continue
		}
		for clientId, conn := range ext.Downstreams {
			if conn.Application.Upstreams[target.Owner.Id] == nil {
				conn.RemoteApplication = target.Owner
				conn.Application.Upstreams[target.Owner.Id] = conn
				target.Owner.Downstreams[clientId] = conn
			}
			delete(conn.Application.Upstreams, extId)
		}
		delete(w.Applications, extId)
	}
}

type appGroup struct {
	app     *model.Application
	members map[model.ApplicationId]*model.Application
}

func (c *Constructor) groupApplications(w *model.World, groups map[model.ApplicationId]*appGroup, project *db.Project) {
	for _, group := range groups {
		categories := utils.NewStringSet()
		for _, app := range group.members {
			for _, svc := range app.KubernetesServices {
				found := false
				for _, existingSvc := range group.app.KubernetesServices {
					if svc.Name == existingSvc.Name && svc.Namespace == existingSvc.Namespace {
						found = true
						break
					}
				}
				if !found {
					group.app.KubernetesServices = append(group.app.KubernetesServices, svc)
				}
				svc.DestinationApps[group.app.Id] = group.app
				delete(svc.DestinationApps, app.Id)
			}
			group.app.DesiredInstances = merge(group.app.DesiredInstances, app.DesiredInstances, timeseries.NanSum)
			for _, instance := range app.Instances {
				instance.Owner = group.app
				instance.ClusterComponent = app
				group.app.Instances = append(group.app.Instances, instance)
			}
			categories.Add(string(app.Category))

			for status, ts := range app.InboundRequestsCount {
				group.app.InboundRequestsCount[status] = merge(group.app.InboundRequestsCount[status], ts, timeseries.NanSum)
			}
			for le, ts := range app.InboundRequestsHistogram {
				group.app.InboundRequestsHistogram[le] = merge(group.app.InboundRequestsHistogram[le], ts, timeseries.NanSum)
			}
			for le, ts := range app.InboundLLMRequestsHistogram {
				group.app.InboundLLMRequestsHistogram[le] = merge(group.app.InboundLLMRequestsHistogram[le], ts, timeseries.NanSum)
			}
			delete(w.Applications, app.Id)
		}
		group.app.Category = model.ApplicationCategory(categories.GetFirst())
		if group.app.Category == "" {
			group.app.Category = project.CalcApplicationCategory(group.app.Id)
		}
	}
}

func (c *Constructor) groupCustomApplications(w *model.World, project *db.Project) {
	customApps := map[model.ApplicationId]*appGroup{}
	for _, app := range w.Applications {
		customName := app.GetAnnotation(model.ApplicationAnnotationCustomName)
		if customName == "" {
			continue
		}
		id := c.newApplicationId(project.ClusterId(), app.Id.Namespace, model.ApplicationKindCustomApplication, customName)
		group := customApps[id]
		if group == nil {
			group = &appGroup{app: w.GetOrCreateApplication(id, true), members: map[model.ApplicationId]*model.Application{}}
			customApps[id] = group
		}
		group.members[app.Id] = app
	}
	c.groupApplications(w, customApps, project)
}

func (c *Constructor) groupInferenceApplications(w *model.World, inferenceGroups map[model.ApplicationId]model.ApplicationId, project *db.Project) {
	groups := map[model.ApplicationId]*appGroup{}
	for deploymentId, id := range inferenceGroups {
		app := w.GetApplication(deploymentId)
		if app == nil {
			continue
		}
		group := groups[id]
		if group == nil {
			group = &appGroup{app: w.GetOrCreateApplication(id, false), members: map[model.ApplicationId]*model.Application{}}
			groups[id] = group
		}
		group.members[app.Id] = app
	}
	c.groupApplications(w, groups, project)
}

func (c *Constructor) joinDBClusterComponents(w *model.World, project *db.Project) {
	dbClusters := map[model.ApplicationId]*appGroup{}
	for _, app := range w.Applications {
		for _, instance := range app.Instances {
			if instance.ClusterName.Value() == "" {
				continue
			}
			id := c.newApplicationId(project.ClusterId(), app.Id.Namespace, model.ApplicationKindDatabaseCluster, instance.ClusterName.Value())
			cluster := dbClusters[id]
			if cluster == nil {
				cluster = &appGroup{app: w.GetOrCreateApplication(id, false), members: map[model.ApplicationId]*model.Application{}}
				dbClusters[id] = cluster
			}
			if app.Cluster.Manager != "" {
				cluster.app.Cluster.Manager = app.Cluster.Manager
			}
			cluster.members[app.Id] = app
		}
	}
	c.groupApplications(w, dbClusters, project)
}

func guessPod(ls model.Labels) string {
	for _, l := range possiblePodLabels {
		if pod := ls[l]; pod != "" {
			return pod
		}
	}
	return ""
}

func guessNamespace(ls model.Labels) string {
	for _, l := range possibleNamespaceLabels {
		if ns := ls[l]; ns != "" {
			return ns
		}
	}
	return ""
}

func findInstance(instancesByPod map[podId]*model.Instance, instancesByListen map[string]*model.Instance, cloudInstancesById map[string]*model.Instance, containers containerCache, m *model.MetricValues, applicationTypes ...model.ApplicationType) *model.Instance {
	ls := m.Labels
	if m.ContainerId != "" {
		if v := containers[m.NodeContainerId]; v.instance != nil {
			return v.instance
		}
	}
	if rdsId := ls["rds_instance_id"]; rdsId != "" {
		return cloudInstancesById[rdsKey(rdsId)]
	}
	if ecId := ls["ec_instance_id"]; ecId != "" {
		return cloudInstancesById[ecKey(ecId)]
	}
	if id := ls["cloudsql_instance_id"]; id != "" {
		return cloudInstancesById[cloudSQLKey(id)]
	}
	if id := ls["memorystore_instance_id"]; id != "" {
		return cloudInstancesById[memorystoreKey(id)]
	}
	if id := ls["oci_db_id"]; id != "" {
		return cloudInstancesById[ociDBKey(id)]
	}
	if id := ls["oci_cache_id"]; id != "" {
		return cloudInstancesById[ociCacheKey(id)]
	}
	address := ls["instance"]
	if ls["address"] != "" {
		address = ls["address"]
	}
	if address != "" {
		if instance := instancesByListen[address]; instance != nil {
			return getActualServiceInstance(freshestSameNameInstance(instance), applicationTypes...)
		}
		if host, _, err := net.SplitHostPort(address); err == nil {
			if instance := instancesByListen[net.JoinHostPort(host, "0")]; instance != nil {
				return getActualServiceInstance(freshestSameNameInstance(instance), applicationTypes...)
			}
		}
	}
	if ns, pod := guessNamespace(ls), guessPod(ls); ns != "" && pod != "" {
		return getActualServiceInstance(freshestSameNameInstance(instancesByPod[podId{name: pod, ns: ns}]), applicationTypes...)
	}
	return nil
}

func freshestSameNameInstance(instance *model.Instance) *model.Instance {
	if instance == nil || !instance.IsObsolete() || instance.Owner == nil {
		return instance
	}
	for _, i := range instance.Owner.Instances {
		if i.Name == instance.Name && !i.IsObsolete() {
			return i
		}
	}
	return instance
}

func getActualServiceInstance(instance *model.Instance, applicationTypes ...model.ApplicationType) *model.Instance {
	if len(applicationTypes) == 0 {
		return instance
	}
	if instance == nil {
		return nil
	}
	for _, t := range applicationTypes {
		if instance.ApplicationTypes()[t] {
			return instance
		}
	}
	return instance
}