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) 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) 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