package reconciliation import ( "errors" "fmt" "sort" "strings" "time" ) const ( containerServiceExternalType = "container-service" containerInstanceExternalType = "container-instance" ) type ContainerObservation struct { SourceID string RuntimeID string Name string Project string Service string ImageDigest string ObservedAt time.Time } type ContainerAlias struct { EntityID string SourceID string RuntimeID string Name string Project string Service string ImageDigest string FirstSeenAt time.Time LastSeenAt time.Time Active bool // TombstonedAt records when the runtime alias stopped being observed. It is // the zero time while the alias is active and is stamped once, so repeated // snapshots keep the original tombstone moment. TombstonedAt time.Time } type ContainerRecreation struct { EntityID string PreviousRuntimeID string CurrentRuntimeID string EventType string EventEntityID string OccurredAt time.Time } type ContainerIdentityResult struct { Aliases []ContainerAlias Current []ContainerAlias Added []string Updated []string Recreated []ContainerRecreation Warnings []string } func ReconcileContainers(observations []ContainerObservation, previous []ContainerAlias, now time.Time) (ContainerIdentityResult, error) { if now.IsZero() { now = time.Now().UTC() } now = now.UTC() records := append([]ContainerAlias(nil), previous...) for i := range records { if records[i].SourceID == "" || records[i].RuntimeID == "" || records[i].EntityID == "" { return ContainerIdentityResult{}, errors.New("invalid previous container alias") } records[i].FirstSeenAt = records[i].FirstSeenAt.UTC() records[i].LastSeenAt = records[i].LastSeenAt.UTC() records[i].TombstonedAt = records[i].TombstonedAt.UTC() records[i].Active = false } exact := make(map[string][]int) stable := make(map[string][]int) for i, record := range records { exact[containerAliasKey(record.SourceID, record.RuntimeID)] = append(exact[containerAliasKey(record.SourceID, record.RuntimeID)], i) if key := containerStableKey(record.SourceID, record.Project, record.Service); key != "" { stable[key] = append(stable[key], i) } } seen := make(map[string]struct{}, len(observations)) result := ContainerIdentityResult{} for _, observation := range observations { if err := validateContainerObservation(observation); err != nil { return ContainerIdentityResult{}, err } runtimeKey := containerAliasKey(observation.SourceID, observation.RuntimeID) if _, ok := seen[runtimeKey]; ok { return ContainerIdentityResult{}, fmt.Errorf("duplicate container observation %q", runtimeKey) } seen[runtimeKey] = struct{}{} var record ContainerAlias if candidates := exact[runtimeKey]; len(candidates) == 1 { record = records[candidates[0]] record.Active = true record.Name = observation.Name record.Project = observation.Project record.Service = observation.Service record.ImageDigest = observation.ImageDigest record.LastSeenAt = observation.ObservedAt.UTC() result.Updated = append(result.Updated, record.EntityID) } else if key := containerStableKey(observation.SourceID, observation.Project, observation.Service); key != "" && len(stable[key]) == 1 { old := records[stable[key][0]] old.Active = false record = ContainerAlias{ EntityID: old.EntityID, SourceID: observation.SourceID, RuntimeID: observation.RuntimeID, Name: observation.Name, Project: observation.Project, Service: observation.Service, ImageDigest: observation.ImageDigest, FirstSeenAt: old.FirstSeenAt, LastSeenAt: observation.ObservedAt.UTC(), Active: true, } records = append(records, record) recreation := ContainerRecreation{ EntityID: record.EntityID, PreviousRuntimeID: old.RuntimeID, CurrentRuntimeID: record.RuntimeID, EventType: "container.recreated", EventEntityID: record.EntityID, OccurredAt: observation.ObservedAt.UTC(), } result.Recreated = append(result.Recreated, recreation) result.Updated = append(result.Updated, record.EntityID) continue } else { externalType := containerInstanceExternalType externalID := observation.RuntimeID if key := containerStableKey(observation.SourceID, observation.Project, observation.Service); key != "" { if len(stable[key]) > 1 { result.Warnings = append(result.Warnings, "AMBIGUOUS_COMPOSE_IDENTITY_NO_MERGE") } else { externalType = containerServiceExternalType externalID = observation.Project + "\x00" + observation.Service } } entityID := StableEntityID(observation.SourceID, externalType, externalID) record = ContainerAlias{ EntityID: entityID, SourceID: observation.SourceID, RuntimeID: observation.RuntimeID, Name: observation.Name, Project: observation.Project, Service: observation.Service, ImageDigest: observation.ImageDigest, FirstSeenAt: observation.ObservedAt.UTC(), LastSeenAt: observation.ObservedAt.UTC(), Active: true, } records = append(records, record) result.Added = append(result.Added, entityID) continue } records = replaceAlias(records, record) } // A runtime alias that is no longer observed is tombstoned rather than // deleted, and keeps the moment it was first found missing. for i := range records { if records[i].Active { records[i].TombstonedAt = time.Time{} } else if records[i].TombstonedAt.IsZero() { records[i].TombstonedAt = now } } sort.Strings(result.Added) sort.Strings(result.Updated) sort.Slice(result.Recreated, func(i, j int) bool { if result.Recreated[i].EntityID != result.Recreated[j].EntityID { return result.Recreated[i].EntityID < result.Recreated[j].EntityID } return result.Recreated[i].CurrentRuntimeID < result.Recreated[j].CurrentRuntimeID }) sort.Strings(result.Warnings) sort.Slice(records, func(i, j int) bool { if records[i].EntityID != records[j].EntityID { return records[i].EntityID < records[j].EntityID } return records[i].RuntimeID < records[j].RuntimeID }) result.Aliases = records for _, record := range records { if record.Active { result.Current = append(result.Current, record) } } return result, nil } func validateContainerObservation(observation ContainerObservation) error { if strings.TrimSpace(observation.SourceID) == "" || strings.TrimSpace(observation.RuntimeID) == "" || strings.TrimSpace(observation.Name) == "" || observation.ObservedAt.IsZero() { return errors.New("invalid container observation") } if (strings.TrimSpace(observation.Project) == "") != (strings.TrimSpace(observation.Service) == "") { return errors.New("compose project and service must be supplied together") } return nil } func containerAliasKey(sourceID, runtimeID string) string { return sourceID + "\x00" + runtimeID } func containerStableKey(sourceID, project, service string) string { if strings.TrimSpace(project) == "" || strings.TrimSpace(service) == "" { return "" } return sourceID + "\x00" + project + "\x00" + service } func replaceAlias(records []ContainerAlias, updated ContainerAlias) []ContainerAlias { for i := range records { if records[i].EntityID == updated.EntityID && records[i].RuntimeID == updated.RuntimeID { records[i] = updated return records } } return append(records, updated) }