constructor/connections.go

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

package constructor

import (
	"net"
	"strings"

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

func (c *Constructor) loadAppToAppConnections(w *model.World, metrics map[string][]*model.MetricValues, fqdn2ip map[string]*utils.StringSet, project *db.Project) {
	for queryName := range metrics {
		if !strings.HasPrefix(queryName, "rr_connection") || strings.HasSuffix(queryName, "_raw") {
			continue
		}
		for _, mv := range metrics[queryName] {
			appId, err := model.NewApplicationIdFromString(mv.Labels["app"], project.ClusterId())
			if err != nil {
				klog.Warningln(err)
				continue
			}
			destId, err := model.NewApplicationIdFromString(mv.Labels["dest"], project.ClusterId())
			if err != nil {
				klog.Warningln(err)
				continue
			}
			app := w.GetOrCreateApplication(appId, false)
			conn := app.Upstreams[destId]
			if conn == nil {
				dest := w.GetOrCreateApplication(destId, false)
				if destId.Kind == model.ApplicationKindExternalService {
					if fqdn, port, _ := net.SplitHostPort(destId.Name); fqdn != "" && port != "" {
						if ips := fqdn2ip[fqdn]; ips != nil {
							for _, ip := range ips.Items() {
								instance := dest.GetOrCreateInstance(ip, nil)
								instance.TcpListens[model.Listen{IP: ip, Port: port}] = true
							}
						}
					}
					if len(dest.Instances) == 0 {
						dest.GetOrCreateInstance(destId.Name, nil)
					}
				}
				conn = &model.AppToAppConnection{
					Application:       app,
					RemoteApplication: dest,
					RequestsCount:     map[model.Protocol]map[string]*timeseries.TimeSeries{},
					RequestsLatency:   map[model.Protocol]*timeseries.TimeSeries{},
					Endpoints:         utils.NewStringSet(),
				}
				app.Upstreams[destId] = conn
				dest.Downstreams[appId] = conn
			}
			switch queryName {
			case qRecordingRuleApplicationTCPSuccessful:
				conn.SuccessfulConnections = merge(conn.SuccessfulConnections, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPActive:
				conn.Active = merge(conn.Active, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPFailed:
				conn.FailedConnections = merge(conn.FailedConnections, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPConnectionTime:
				conn.ConnectionTime = merge(conn.ConnectionTime, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPBytesSent:
				conn.BytesSent = merge(conn.BytesSent, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPBytesReceived:
				conn.BytesReceived = merge(conn.BytesReceived, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationTCPRetransmissions:
				conn.Retransmissions = merge(conn.Retransmissions, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationNetLatency:
				conn.Rtt = merge(conn.Rtt, mv.Values, timeseries.Any)
			case qRecordingRuleApplicationL7Latency:
				proto := model.Protocol(mv.Labels["proto"])
				conn.RequestsLatency[proto] = merge(conn.RequestsLatency[proto], mv.Values, timeseries.Any)
			case qRecordingRuleApplicationL7Requests:
				proto := model.Protocol(mv.Labels["proto"])
				switch proto {
				case model.ProtocolRabbitmq, model.ProtocolNats:
					proto += model.Protocol("-" + mv.Labels["method"])
				}
				if conn.RequestsCount[proto] == nil {
					conn.RequestsCount[proto] = map[string]*timeseries.TimeSeries{}
				}
				status := mv.Labels["status"]
				conn.RequestsCount[proto][status] = merge(conn.RequestsCount[proto][status], mv.Values, timeseries.NanSum)
			case qRecordingRuleApplicationExternalEndpoint:
				if mv.ActualDestination != "" && mv.ActualDestination != ":" { // failed-only connections have no actual destination
					conn.Endpoints.Add(mv.ActualDestination)
					conn.RemoteApplication.GetOrCreateInstance(mv.ActualDestination, nil)
					if ip, port, _ := net.SplitHostPort(mv.ActualDestination); ip != "" && port != "" {
						if net.ParseIP(ip) != nil {
							instance := conn.RemoteApplication.GetOrCreateInstance(mv.ActualDestination, nil)
							instance.TcpListens[model.Listen{IP: ip, Port: port}] = true
						}
					}
				}
			}
		}
	}
}