output

type BrokerWriter ¶

type BrokerWriter struct {
	// contains filtered or unexported fields
}

func NewBrokerWriter(br broker.Handler) *BrokerWriter

func (s *BrokerWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	config config.PipelineConfig,
) error

type CompositeWriter ¶

type CompositeWriter struct {
	// contains filtered or unexported fields
}

a wrapper which allows to have multiple writers under one entity

func NewCompositeWriter(writer ...Writer) *CompositeWriter

func (w *CompositeWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	config config.PipelineConfig,
) error

type ContentDeduplicatorWriter ¶ added in v1.0.2

type ContentDeduplicatorWriter struct {
	// contains filtered or unexported fields
}

ContentDeduplicatorWriter is a streaming output wrapper that suppresses byte-identical republishes of the SAME resource on the broker path.

It keys resources by KubernetesResourceMeta.UID and remembers a sha256 hash of the last payload published for that UID. On ADD/UPDATE, if the incoming payload hashes to the same value already recorded for the UID, the write is skipped; otherwise it is forwarded and the recorded hash is updated. This is a bandwidth/DB-write optimisation on top of the resourceVersion-based suppression already performed in the informer UpdateFunc: two distinct resourceVersions can still carry identical wire content (e.g. status churn that normalises away), and only a content hash catches those.

The full object is always forwarded when it IS published: this wrapper never rewrites the wire format into a delta/patch, because Meshery Server consumes full objects from these subjects.

Invariants that keep this safe:

• DELETE is ALWAYS forwarded and evicts the UID from the map. Eviction bounds the map to the set of currently-live UIDs and guarantees that a re-created object (same UID reused, or a new UID) republishes fresh.
• Resources with an empty/absent UID are never deduplicated; each is forwarded as-is (mirrors InMemoryDeduplicatorStreamingWriter).
• Ordering and event semantics are preserved: this is a pass-through filter, it never reorders, batches, or defers events.

NOTE on informer resyncs: the broker writer (and therefore this wrapper) persists across informer resyncs - a resync recreates the informer factory but not the output writer - so the hash map survives a resync. A resync re-lists every object, and unchanged objects would be suppressed here. That is acceptable while Meshery Server still holds those unchanged objects, but because a resync is also a recovery path it must be an explicit opt-in (see config.EnvBrokerContentDedup); it is OFF by default so the default behaviour - republish everything - is unchanged.

func NewContentDeduplicatorWriter(realWriter Writer) *ContentDeduplicatorWriter

func (w *ContentDeduplicatorWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	cfg config.PipelineConfig,
) error

type FileWriter ¶

type FileWriter struct {
	// contains filtered or unexported fields
}

func NewFileWriter(fw file.Writer) *FileWriter

func (s *FileWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	config config.PipelineConfig,
) error

type InMemoryDeduplicatorStreamingWriter ¶ added in v0.8.23

type InMemoryDeduplicatorStreamingWriter struct {
	// contains filtered or unexported fields
}

InMemoryDeduplicatorStreamingWriter writes each unique resource once immediately upon first seeing it

func NewInMemoryDeduplicatorStreamingWriter(realWriter Writer) *InMemoryDeduplicatorStreamingWriter

func (w *InMemoryDeduplicatorStreamingWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	cfg config.PipelineConfig,
) error

type InMemoryDeduplicatorWriter ¶

type InMemoryDeduplicatorWriter struct {
	// contains filtered or unexported fields
}

instead of direct write to output destination InMemoryDeduplicatorWriter collects data in memory identifying entity by metadata.uid and write to output only on program exit

func NewInMemoryDeduplicatorWriter(realWritter Writer) *InMemoryDeduplicatorWriter

func (w *InMemoryDeduplicatorWriter) Flush() error

func (w *InMemoryDeduplicatorWriter) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	config config.PipelineConfig,
) error

type Processor ¶

type Processor struct {
	// contains filtered or unexported fields
}

func NewProcessor() *Processor

func (p *Processor) SetOutput(output Writer)

func (p *Processor) Write(
	obj model.KubernetesResource,
	evtype broker.EventType,
	config config.PipelineConfig,
) error

type Writer ¶

type Writer interface {
	Write(
		obj model.KubernetesResource,
		evtype broker.EventType,
		config config.PipelineConfig,
	) error
}