Documentation
¶
Overview ¶
Package datasource holds cross-backend datasource decorators. The Guard is the authorization + audit enforcement seam: it wraps any api.KafkaDataSource and interposes on state-changing operations.
CONTRACT for future features: the Guard EMBEDS api.KafkaDataSource, so read and analytical methods pass through automatically. Every NEW mutating method added to api.KafkaDataSource MUST be overridden here with a gate check + audit record (use the do/doValue helpers and declare its authz resource+action). Denial happens BEFORE any effect. See CLAUDE.md ("Authorization/Audit seam").
Index ¶
- type Guard
- func (g *Guard) AlterBrokerConfig(brokerID int32, key, value string) error
- func (g *Guard) AlterClientQuotas(entity api.ClientQuotaEntity, quotas map[string]float64) error
- func (g *Guard) AlterReplicaLogDir(brokerID int32, topic string, partition int32, logDir string) error
- func (g *Guard) ChangeReplicationFactor(name string, newFactor int16) error
- func (g *Guard) CreateACL(entry api.ACLEntry) error
- func (g *Guard) CreateConnector(connect, name string, config map[string]string) (api.Connector, error)
- func (g *Guard) CreateTopic(name string, numPartitions int32, replicationFactor int16, ...) error
- func (g *Guard) DeleteACL(entry api.ACLEntry) error
- func (g *Guard) DeleteConnector(connect, name string) error
- func (g *Guard) DeleteConsumerGroup(groupID string) error
- func (g *Guard) DeleteConsumerGroupOffsets(groupID string, topic string) error
- func (g *Guard) DeleteSchemaVersion(subject string, version int, permanent bool) error
- func (g *Guard) DeleteSubject(subject string, permanent bool) ([]int, error)
- func (g *Guard) DeleteTopic(name string) error
- func (g *Guard) ExecuteKsql(ctx context.Context, sql string, props map[string]string) (<-chan api.KsqlResultTable, error)
- func (g *Guard) GetConsumerGroups() ([]api.ConsumerGroup, error)
- func (g *Guard) GetSchemaDetails(subjects []string) ([]api.Schema, error)
- func (g *Guard) GetSchemas() ([]api.Schema, error)
- func (g *Guard) GetTopicNames() ([]string, error)
- func (g *Guard) GetTopics() (map[string]api.Topic, error)
- func (g *Guard) IncreasePartitions(name string, totalCount int32) error
- func (g *Guard) PauseConnector(connect, name string) error
- func (g *Guard) ProduceMessage(ctx context.Context, topic string, rec api.ProduceRecord) error
- func (g *Guard) PurgeTopicMessages(name string, partition int32) error
- func (g *Guard) RecreateTopic(name string) error
- func (g *Guard) RegisterSchema(subject, schemaText, schemaType string) (api.Schema, error)
- func (g *Guard) Reload(c appconfig.Config) error
- func (g *Guard) ResetConnectorOffsets(connect, name string) error
- func (g *Guard) ResetConsumerGroupOffsets(ctx context.Context, req api.OffsetResetRequest) error
- func (g *Guard) RestartConnector(connect, name string) error
- func (g *Guard) RestartConnectorTask(connect, name string, taskID int) error
- func (g *Guard) ResumeConnector(connect, name string) error
- func (g *Guard) SetContext(name string) error
- func (g *Guard) SetGlobalCompatibility(level api.CompatibilityLevel) error
- func (g *Guard) SetSubjectCompatibility(subject string, level api.CompatibilityLevel) error
- func (g *Guard) StopConnector(connect, name string) error
- func (g *Guard) UpdateConnectorConfig(connect, name string, config map[string]string) (api.Connector, error)
- func (g *Guard) UpdateTopicConfig(name string, entries map[string]*string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Guard ¶
type Guard struct {
api.KafkaDataSource
// contains filtered or unexported fields
}
Guard wraps a KafkaDataSource, enforcing the authorization Gate and emitting audit records for state-changing operations. It satisfies api.KafkaDataSource by embedding the wrapped implementation.
func NewGuard ¶
NewGuard wraps inner with gate + audit enforcement. It seeds the gate with the inner datasource's current cluster so the first checks resolve the right profile. A nil gate/audit degrades to allow-all / no-op respectively.
func (*Guard) AlterBrokerConfig ¶
func (*Guard) AlterClientQuotas ¶
func (*Guard) AlterReplicaLogDir ¶
func (*Guard) ChangeReplicationFactor ¶
func (*Guard) CreateConnector ¶
func (*Guard) CreateTopic ¶
func (*Guard) DeleteConnector ¶
func (*Guard) DeleteConsumerGroup ¶
func (*Guard) DeleteConsumerGroupOffsets ¶
func (*Guard) DeleteSchemaVersion ¶
func (*Guard) DeleteSubject ¶
func (*Guard) DeleteTopic ¶
func (*Guard) ExecuteKsql ¶
func (*Guard) GetConsumerGroups ¶
func (g *Guard) GetConsumerGroups() ([]api.ConsumerGroup, error)
func (*Guard) GetSchemaDetails ¶
func (*Guard) GetTopicNames ¶
func (*Guard) IncreasePartitions ¶
func (*Guard) PauseConnector ¶
func (*Guard) ProduceMessage ¶
func (*Guard) PurgeTopicMessages ¶
func (*Guard) RecreateTopic ¶
func (*Guard) RegisterSchema ¶
func (*Guard) Reload ¶ added in v0.1.36
Reload forwards an in-place config reload to the wrapped datasource when it supports one (kafds does, the mock does not), then re-resolves the gate's active cluster, which the reload may have changed. Reload is not part of api.KafkaDataSource, so without this method the embedding would hide it and the UI's type assertion for it would always fail. It only changes the in-memory cluster list (nothing is written to Kafka), so it is not gated.
func (*Guard) ResetConnectorOffsets ¶
func (*Guard) ResetConsumerGroupOffsets ¶
func (*Guard) RestartConnector ¶
func (*Guard) RestartConnectorTask ¶
func (*Guard) ResumeConnector ¶
func (*Guard) SetContext ¶
SetContext delegates then re-resolves the gate's active cluster profile.
func (*Guard) SetGlobalCompatibility ¶
func (g *Guard) SetGlobalCompatibility(level api.CompatibilityLevel) error
func (*Guard) SetSubjectCompatibility ¶
func (g *Guard) SetSubjectCompatibility(subject string, level api.CompatibilityLevel) error
Source Files
Directories