Kafka Messages

If kafka logging is configured (see config-full.toml example in /scripts/distro-files), a number of JSON messages are produced to kafka when mutations are made to a subset of datatypes. This page documents the JSON messages using the Go types they represent. Many mutations generate two messages (not counting any changes in synced data instances): the main mutation event and a final "complete" message that is produced when the mutation finishes without error.

labelmap

POST /merge

{
  "Action":     "merge",
  "Target":     uint64,
  "Labels":     []uint64,
  "UUID":       string(versionuuid),
  "MutationID": uint64,
  "Timestamp":  time.Now().String()
}

{
  "Action":     "merge-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String()
}

POST /cleave

{
  "Action":             "cleave",
  "OrigLabel":          uint64,
  "CleavedLabel":       uint64,
  "CleavedSupervoxels": []uint64,
  "MutationID":         uint64,,
  "UUID":               string(versionuuid),
  "Timestamp":          time.Now().String(),
}

{
  "Action":     "cleave-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
}

POST /split-supervoxel

{
  "Action":           "split-supervoxel",
  "Supervoxel":       uint64,
  "SplitSupervoxel":  uint64,
  "RemainSupervoxel": uint64,
  "Split":            string,  // reference to data in blob store
  "MutationID":       uint64,
  "UUID":             string(versionuuid),
  "Timestamp":        time.Now().String()
}

{
  "Action":     "split-supervoxel-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String()
}

POST /split

{
  "Action":     "split",
  "Target":     uint64,
  "NewLabel":   uint64,
  "Split":      string,
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "SVSplits":   map[uint64]SVSplit,
  "Timestamp":  time.Now().String()
}

SVSplit above is the following:
{
  "Split": uint64,
  "Remain": uint64
}

{
  "Action":     "split-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
}

POST /maxlabel

{				
  "Action":    "post-maxlabel",
  "MaxLabel":  uint64,
  "UUID":      string(versionuuid),
  "Timestamp": time.Now().String()
}

POST /nextlabel

{
  "Action":      "post-nextlabel",
  "Start Label": uint64,
  "End Label":   uint64,
  "UUID":        string(versionuuid),
  "Timestamp":   time.Now().String(),
}

labelarray

POST /merge

{
  "Action":     "merge",
  "Target":     uint64,
  "Labels":     []uint64,
  "UUID":       string(versionuuid),
  "MutationID": uint64,
  "Timestamp":  time.Now().String()
}

{
  "Action":     "merge-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String()
}

POST /split

{
  "Action":     "split",
  "Target":     uint64,
  "NewLabel":   uint64,
  "Split":      string,
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String()
}

{
  "Action":     "split-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
}

POST /split-coarse

{
  "Action":     "splitcoarse",
  "Target":     uint64,
  "NewLabel":   uint64,
  "Split":      string,
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String()
}

{
  "Action":     "splitcoarse-complete",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
}

keyvalue

POST /key

POST /keyvalues

{
  "Action":    "postkv",
  "Key":       string,
  "Bytes":     len(data),
  "UUID":      string(uuid),
  "Timestamp": time.Now().String()
}

annotation

Delta JSON structures correspond to the DeltaModifyElements type in Go:

// DeltaModifyElements is a change in the elements assigned to a label.
// Need positions of elements because subscribers may have ROI filtering.
type DeltaModifyElements struct {
	Add []ElementPos
	Del []ElementPos
}

// ElementPos describes the label and kind of an annotation, useful for synchronizing
// changes in data to other data types like labelsz.
type ElementPos struct {
	Label uint64
	Kind  uint8
	Pos   [3]uint32
}

If the "Delta" field is too large for a Kafka message, a "DataRef" with string value is written. The DataRef is the key for the full data in the DVID blobstore API.

Merge in synced labelmap, labelarray, or labelvol instance.

{
  "Action":     "merge",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
  "Delta":      DeltaModifyElements,
}

Cleave in synced labelmap, labelarray, or labelvol instance.

{
  "Action":     "cleave",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
  "Delta":      DeltaModifyElements,
}

Coarse split in synced labelmap, labelarray, or labelvol instance.

{
  "Action":     "split-coarse",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
  "Delta":      DeltaModifyElements,
}

Arbitrary split in synced labelmap, labelarray, or labelvol instance.

{
  "Action":     "split",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
  "Delta":      DeltaModifyElements,
}

Changes due to block ingestion in synced labelmap, labelarray, or labelblk instance.

{
  "Action":    "ingest-block",
  "UUID":      string(versionuuid),
  "Timestamp": time.Now().String(),
  "Delta":     DeltaModifyElements,
}

Changes due to block mutation in synced labelmap, labelarray, or labelblk instance.

{
  "Action":     "mutate-block",
  "MutationID": uint64,
  "UUID":       string(versionuuid),
  "Timestamp":  time.Now().String(),
  "Delta":      DeltaModifyElements,
}