Skip to content

fix(database): create alert events/metas with owner on MySQL #4507

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pepordev wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(database): create alert events/metas with owner on MySQL #4507

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
201 changes: 109 additions & 92 deletions pkg/database/alerts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,109 +436,123 @@ func parseAlertTimes(alert *models.Alert, logger log.FieldLogger) (time.Time, ti
return start, stop
}

func buildEventCreates(ctx context.Context, logger log.FieldLogger, client *ent.Client, machineID string, alertItem *models.Alert) ([]*ent.Event, error) {
// let's track when we strip or drop data, notify outside of loop to avoid spam
func marshalEventSerialized(logger log.FieldLogger, machineID, scenario string, eventItem *models.Event) (string, bool, bool, error) {
stripped := false
dropped := false

if len(alertItem.Events) == 0 {
return nil, nil
marshallMetas, err := json.Marshal(eventItem.Meta)
if err != nil {
return "", stripped, dropped, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail)
}

eventBulk := make([]*ent.EventCreate, len(alertItem.Events))

for i, eventItem := range alertItem.Events {
ts, err := time.Parse(time.RFC3339, *eventItem.Timestamp)
if err != nil {
logger.Errorf("creating alert: Failed to parse event timestamp '%s', defaulting to now: %s", *eventItem.Timestamp, err)

ts = time.Now().UTC()
}
// the serialized field is too big, let's try to progressively strip it
if event.SerializedValidator(string(marshallMetas)) != nil {
stripped = true

marshallMetas, err := json.Marshal(eventItem.Meta)
if err != nil {
return nil, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail)
}

// the serialized field is too big, let's try to progressively strip it
if event.SerializedValidator(string(marshallMetas)) != nil {
stripped = true

valid := false
stripSize := 2048

for !valid && stripSize > 0 {
for _, serializedItem := range eventItem.Meta {
if len(serializedItem.Value) > stripSize*2 {
serializedItem.Value = serializedItem.Value[:stripSize] + "<stripped>"
}
}

marshallMetas, err = json.Marshal(eventItem.Meta)
if err != nil {
return nil, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail)
}
valid := false
stripSize := 2048

if event.SerializedValidator(string(marshallMetas)) == nil {
valid = true
for !valid && stripSize > 0 {
for _, serializedItem := range eventItem.Meta {
if len(serializedItem.Value) > stripSize*2 {
serializedItem.Value = serializedItem.Value[:stripSize] + "<stripped>"
}
}

stripSize /= 2
marshallMetas, err = json.Marshal(eventItem.Meta)
if err != nil {
return "", stripped, dropped, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail)
}

// nothing worked, drop it
if !valid {
dropped = true
stripped = false
marshallMetas = []byte("")
if event.SerializedValidator(string(marshallMetas)) == nil {
valid = true
}

stripSize /= 2
}

eventBulk[i] = client.Event.Create().
SetTime(ts).
SetSerialized(string(marshallMetas))
// nothing worked, drop it
if !valid {
dropped = true
stripped = false
marshallMetas = []byte("")
}
}

if stripped {
logger.Warningf("stripped 'serialized' field (machine %s / scenario %s)", machineID, *alertItem.Scenario)
logger.Warningf("stripped 'serialized' field (machine %s / scenario %s)", machineID, scenario)
}

if dropped {
logger.Warningf("dropped 'serialized' field (machine %s / scenario %s)", machineID, *alertItem.Scenario)
logger.Warningf("dropped 'serialized' field (machine %s / scenario %s)", machineID, scenario)
}

return client.Event.CreateBulk(eventBulk...).Save(ctx)
return string(marshallMetas), stripped, dropped, nil
}

func buildMetaCreates(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertItem *models.Alert) ([]*ent.Meta, error) {
if len(alertItem.Meta) == 0 {
return nil, nil
// createAlertEventsForAlert inserts events with owner set at creation time.
// Pre-creating events then linking via AddEvents fails on MySQL (alert_events constraint).
func createAlertEventsForAlert(ctx context.Context, logger log.FieldLogger, client *ent.Client, machineID, scenario string, alertID int, events []*models.Event) error {
for _, eventItem := range events {
ts, err := time.Parse(time.RFC3339, *eventItem.Timestamp)
if err != nil {
logger.Errorf("creating alert: Failed to parse event timestamp '%s', defaulting to now: %s", *eventItem.Timestamp, err)

ts = time.Now().UTC()
}

serialized, _, _, err := marshalEventSerialized(logger, machineID, scenario, eventItem)
if err != nil {
return err
}

if _, err := client.Event.Create().
SetTime(ts).
SetSerialized(serialized).
SetOwnerID(alertID).
Save(ctx); err != nil {
return fmt.Errorf("creating alert event: %w", err)
}
}

metaBulk := make([]*ent.MetaCreate, len(alertItem.Meta))
return nil
}

for i, metaItem := range alertItem.Meta {
key := metaItem.Key
value := metaItem.Value
func normalizeMetaItem(logger log.FieldLogger, metaItem *models.MetaItems0) (string, string) {
key := metaItem.Key
value := metaItem.Value

if len(metaItem.Value) > 4095 {
logger.Warningf("truncated meta %s: value too long", metaItem.Key)
if len(metaItem.Value) > 4095 {
logger.Warningf("truncated meta %s: value too long", metaItem.Key)

value = value[:4095]
}
value = value[:4095]
}

if len(metaItem.Key) > 255 {
logger.Warningf("truncated meta %s: key too long", metaItem.Key)
if len(metaItem.Key) > 255 {
logger.Warningf("truncated meta %s: key too long", metaItem.Key)

key = key[:255]
}
key = key[:255]
}

return key, value
}

// createAlertMetasForAlert inserts alert-level metas with owner set at creation time.
// Pre-creating metas then linking via AddMetas fails on MySQL (alert_metas constraint).
func createAlertMetasForAlert(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertID int, metaItems []*models.MetaItems0) error {
for _, metaItem := range metaItems {
key, value := normalizeMetaItem(logger, metaItem)

metaBulk[i] = client.Meta.Create().
if _, err := client.Meta.Create().
SetKey(key).
SetValue(value)
SetValue(value).
SetOwnerID(alertID).
Save(ctx); err != nil {
return fmt.Errorf("creating alert meta %s: %w", key, err)
}
}

return client.Meta.CreateBulk(metaBulk...).Save(ctx)
return nil
}

func (c *Client) buildDecisions(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertItem *models.Alert, stopAtTime time.Time) ([]*ent.Decision, int, error) {
Expand Down Expand Up @@ -588,25 +602,32 @@ func (c *Client) saveAlerts(ctx context.Context, client *ent.Client, batch []ale
return nil, nil
}

// extract builders in the same order
builders := make([]*ent.AlertCreate, len(batch))
// Insert alerts one-by-one; events and metas are created after save with owner set.
ret := make([]string, len(batch))
for i := range batch {
if batch[i].builder == nil {
return nil, fmt.Errorf("nil alert builder at index %d", i)
}

builders[i] = batch[i].builder
}

alertsCreateBulk, err := client.Alert.CreateBulk(builders...).Save(ctx)
if err != nil {
return nil, fmt.Errorf("bulk creating alert: %w: %w", err, BulkError)
}
a, err := batch[i].builder.Save(ctx)
if err != nil {
return nil, fmt.Errorf("creating alert: %w: %w", err, BulkError)
}

ret := make([]string, len(alertsCreateBulk))
for i, a := range alertsCreateBulk {
ret[i] = strconv.Itoa(a.ID)

if len(batch[i].events) > 0 {
if err := createAlertEventsForAlert(ctx, c.Log, client, batch[i].machineID, batch[i].scenario, a.ID, batch[i].events); err != nil {
return nil, fmt.Errorf("creating events for alert %d: %w: %w", a.ID, err, BulkError)
}
}

if len(batch[i].meta) > 0 {
if err := createAlertMetasForAlert(ctx, c.Log, client, a.ID, batch[i].meta); err != nil {
return nil, fmt.Errorf("creating metas for alert %d: %w: %w", a.ID, err, BulkError)
}
}

d := batch[i].decisions
if len(d) == 0 {
continue
Expand All @@ -627,6 +648,10 @@ func (c *Client) saveAlerts(ctx context.Context, client *ent.Client, batch []ale

type alertCreatePlan struct {
builder *ent.AlertCreate
machineID string
scenario string
events []*models.Event
meta []*models.MetaItems0
decisions []*ent.Decision
}

Expand All @@ -648,16 +673,6 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner *
c.Log.Info(disp)
}

events, err := buildEventCreates(ctx, c.Log, txEnt, machineID, alertItem)
if err != nil {
return nil, rollbackOnError(tx, err, fmt.Sprintf("building events for alert %s", alertItem.UUID))
}

metas, err := buildMetaCreates(ctx, c.Log, txEnt, alertItem)
if err != nil {
c.Log.Warningf("error creating alert meta: %s", err)
}

decisions, discardCount, err := c.buildDecisions(ctx, c.Log, txEnt, alertItem, stopAtTime)
if err != nil {
return nil, rollbackOnError(tx, err, fmt.Sprintf("building decisions for alert %s", alertItem.UUID))
Expand Down Expand Up @@ -692,16 +707,18 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner *
SetScenarioHash(*alertItem.ScenarioHash).
SetRemediation(alertItem.Remediation).
SetUUID(alertItem.UUID).
SetKind(alertItem.Kind).
AddEvents(events...).
AddMetas(metas...)
SetKind(alertItem.Kind)

if owner != nil {
builder.SetOwner(owner)
}

batch = append(batch, alertCreatePlan{
builder: builder,
machineID: machineID,
scenario: *alertItem.Scenario,
events: alertItem.Events,
meta: alertItem.Meta,
decisions: decisions,
})
}
Expand Down
Loading
Loading