ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
104 lines
3.7 KiB
Go
104 lines
3.7 KiB
Go
package reconciliation
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/itworx/pulse/internal/datasource"
|
|
)
|
|
|
|
type Observation struct {
|
|
SourceID, ExternalType, ExternalID, EntityType, CanonicalName, DisplayName string
|
|
ObservedAt time.Time
|
|
}
|
|
|
|
type Existing struct {
|
|
EntityID string
|
|
SourceID, ExternalType, ExternalID string
|
|
CanonicalName, DisplayName string
|
|
Tombstoned bool
|
|
// TombstonedAt records when the entity was tombstoned. It is the zero time
|
|
// while the entity is live and maps onto entities.tombstoned_at.
|
|
TombstonedAt time.Time
|
|
}
|
|
type Override struct{ EntityID, FieldName, Value string }
|
|
type Result struct {
|
|
Entities []Existing
|
|
Added, Updated, Tombstoned []string
|
|
Aliases map[string]string
|
|
Warnings []string
|
|
}
|
|
|
|
func StableEntityID(sourceID, externalType, externalID string) string {
|
|
hash := sha256.Sum256([]byte("itworx-pulse/entity/v1/" + sourceID + "\x00" + externalType + "\x00" + externalID))
|
|
b := hash[:16]
|
|
b[6] = (b[6] & 0x0f) | 0x50
|
|
b[8] = (b[8] & 0x3f) | 0x80
|
|
return fmt.Sprintf("%s-%s-%s-%s-%s", hex.EncodeToString(b[0:4]), hex.EncodeToString(b[4:6]), hex.EncodeToString(b[6:8]), hex.EncodeToString(b[8:10]), hex.EncodeToString(b[10:16]))
|
|
}
|
|
|
|
func Reconcile(sourceID string, health datasource.HealthState, observations []Observation, previous []Existing, overrides []Override, now time.Time) (Result, error) {
|
|
if strings.TrimSpace(sourceID) == "" {
|
|
return Result{}, errors.New("source id is required")
|
|
}
|
|
if health == datasource.HealthUnknown || health == datasource.HealthDisabled {
|
|
return Result{Entities: previous, Warnings: []string{"SOURCE_NOT_HEALTHY_NO_TOMBSTONES"}}, nil
|
|
}
|
|
seen := make(map[string]struct{}, len(observations))
|
|
result := Result{Aliases: make(map[string]string)}
|
|
overrideMap := make(map[string]string)
|
|
for _, o := range overrides {
|
|
overrideMap[o.EntityID+"\x00"+o.FieldName] = o.Value
|
|
}
|
|
previousByAlias := make(map[string]Existing)
|
|
for _, e := range previous {
|
|
previousByAlias[e.SourceID+"\x00"+e.ExternalType+"\x00"+e.ExternalID] = e
|
|
}
|
|
for _, o := range observations {
|
|
if o.SourceID != sourceID || o.ExternalType == "" || o.ExternalID == "" || o.EntityType == "" || o.CanonicalName == "" || o.ObservedAt.IsZero() {
|
|
return Result{}, errors.New("invalid observation")
|
|
}
|
|
key := o.SourceID + "\x00" + o.ExternalType + "\x00" + o.ExternalID
|
|
if _, ok := seen[key]; ok {
|
|
return Result{}, fmt.Errorf("duplicate observation %q", key)
|
|
}
|
|
seen[key] = struct{}{}
|
|
id := StableEntityID(o.SourceID, o.ExternalType, o.ExternalID)
|
|
if old, ok := previousByAlias[key]; ok {
|
|
id = old.EntityID
|
|
}
|
|
display := o.DisplayName
|
|
if value, ok := overrideMap[id+"\x00displayName"]; ok {
|
|
display = value
|
|
}
|
|
entity := Existing{EntityID: id, SourceID: o.SourceID, ExternalType: o.ExternalType, ExternalID: o.ExternalID, CanonicalName: o.CanonicalName, DisplayName: display}
|
|
result.Entities = append(result.Entities, entity)
|
|
result.Aliases[key] = id
|
|
if _, ok := previousByAlias[key]; ok {
|
|
result.Updated = append(result.Updated, id)
|
|
} else {
|
|
result.Added = append(result.Added, id)
|
|
}
|
|
}
|
|
if now.IsZero() {
|
|
now = time.Now().UTC()
|
|
}
|
|
now = now.UTC()
|
|
for _, old := range previous {
|
|
key := old.SourceID + "\x00" + old.ExternalType + "\x00" + old.ExternalID
|
|
if old.SourceID == sourceID {
|
|
if _, ok := seen[key]; !ok && !old.Tombstoned {
|
|
old.Tombstoned = true
|
|
old.TombstonedAt = now
|
|
result.Entities = append(result.Entities, old)
|
|
result.Tombstoned = append(result.Tombstoned, old.EntityID)
|
|
}
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|