hubcio commented on code in PR #4716: URL: https://github.com/apache/iggy/pull/4716#discussion_r4469991922
########## examples/go/multi-tenant/consumer/main.go: ########## @@ -0,0 +1,325 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package main reads every tenant's stream as that tenant, with the tenant's +// own restricted credentials. +// +// The streams and their users come from the multi-tenant producer, so run it +// first. The consumer only adds its own read-only user per tenant, then proves +// that each of those users can read its own stream and no other. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "strconv" + "time" + + "github.com/apache/iggy/examples/go/common" + "github.com/apache/iggy/foreign/go/client" + "github.com/apache/iggy/foreign/go/client/tcp" + iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" +) + +const ( + tenantPassword = "secret" + topicName = "events" + partitionID = uint32(0) + messagesPerBatch = 5 + defaultTenants = 3 + consumerUsername = "tenant_%d_consumer" + tenantStream = "tenant_%d_stream" + connectTimeout = 30 * time.Second + pollAttempts = 20 + pollRetryInterval = 100 * time.Millisecond +) + +type tenant struct { + id int + stream string + user string +} + +func main() { + serverAddr, tcpOptions := getTcpOptions() + tenantsCount := tenantsFromEnv() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + log.Printf("Multi-tenant consumer started, tenants: %d", tenantsCount) + + root, err := dial(ctx, serverAddr, tcpOptions, common.DefaultRootUsername, common.DefaultRootPassword) + if err != nil { + log.Fatalf("Connect as the root user: %v", err) + } + + tenants := make([]tenant, 0, tenantsCount) + for index := 1; index <= tenantsCount; index++ { + current := tenant{ + id: index, + stream: fmt.Sprintf(tenantStream, index), + user: fmt.Sprintf(consumerUsername, index), + } + stream, err := existingStream(ctx, root, current.stream) + if err != nil { + closeClient(root) + log.Fatalf("Tenant %d: %v", current.id, err) + } + topic, err := existingTopic(ctx, root, stream.Id, topicName) + if err != nil { + closeClient(root) + log.Fatalf("Tenant %d: %v", current.id, err) + } + if err := ensureUser(ctx, root, current.user, consumerPermissions(stream.Id, topic.Id)); err != nil { + closeClient(root) + log.Fatalf("Prepare user %q: %v", current.user, err) + } + tenants = append(tenants, current) + } + closeClient(root) + + for _, current := range tenants { + consumer, err := dial(ctx, serverAddr, tcpOptions, current.user, tenantPassword) + if err != nil { + log.Fatalf("Connect as %q: %v", current.user, err) + } + received, err := receiveMessages(ctx, consumer, current) + if err != nil { + closeClient(consumer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + if err := ensureNoAccess(ctx, consumer, current, tenants); err != nil { + closeClient(consumer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + log.Printf("Tenant %d (%s) read %d message(s) from its own stream %q", + current.id, current.user, received, current.stream) + closeClient(consumer) + } + + log.Printf("Multi-tenant consumer finished, %d tenant(s) read their own stream only", len(tenants)) +} + +// receiveMessages polls the tenant's own topic from the first offset, with the +// tenant's own client. +func receiveMessages(ctx context.Context, cli iggcon.Client, current tenant) (int, error) { + streamIdentifier, err := iggcon.NewIdentifier(current.stream) + if err != nil { + return 0, err + } + topicIdentifier, err := iggcon.NewIdentifier(topicName) + if err != nil { + return 0, err + } + + options := iggcon.DefaultPollOptions() + options.Count = messagesPerBatch + + for attempt := 1; ; attempt++ { + polled, err := cli.PollMessages(ctx, streamIdentifier, topicIdentifier, partitionID, + iggcon.DefaultConsumer(), iggcon.OffsetPollingStrategy(0), options) + if err != nil { + return 0, fmt.Errorf("poll stream %q: %w", current.stream, err) + } + if len(polled.Messages) == 0 { + if attempt >= pollAttempts { + return 0, fmt.Errorf("no messages in stream %q after %d polls: run the multi-tenant producer first", + current.stream, pollAttempts) + } + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-time.After(pollRetryInterval): + } + continue + } + for _, message := range polled.Messages { + log.Printf("Tenant %d (%s) received %q from %s at offset %d", + current.id, current.user, string(message.Payload), current.stream, message.Header.Offset) + } + return len(polled.Messages), nil + } +} + +// ensureNoAccess proves the tenant can read its own stream and cannot read any +// other tenant's stream. +func ensureNoAccess(ctx context.Context, cli iggcon.Client, current tenant, tenants []tenant) error { Review Comment: simplification: the consumer duplicates the producer's access checks, user setup, connection helpers, and configuration parsing. share these implementations through the existing `examples/go/common` package. ########## examples/go/multi-tenant/producer/main.go: ########## @@ -0,0 +1,325 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package main shows one stream and one restricted user per tenant. +// +// The root user creates every stream and user, then each tenant connects with +// its own credentials and sends to its own stream only. The last step of every +// tenant is an attempt to read another tenant's stream, which the server must +// refuse. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "strconv" + "time" + + "github.com/apache/iggy/examples/go/common" + "github.com/apache/iggy/foreign/go/client" + "github.com/apache/iggy/foreign/go/client/tcp" + iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" +) + +const ( + tenantPassword = "secret" + topicName = "events" + partitionID = uint32(0) + partitionsCount = uint32(1) + messagesPerBatch = 5 + batchesPerTenant = 2 + defaultTenants = 3 + producerUsername = "tenant_%d_producer" + tenantStream = "tenant_%d_stream" + connectTimeout = 30 * time.Second +) + +type tenant struct { + id int + stream string + user string +} + +func main() { + serverAddr, tcpOptions := getTcpOptions() + tenantsCount := tenantsFromEnv() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + log.Printf("Multi-tenant producer started, tenants: %d", tenantsCount) + + root, err := dial(ctx, serverAddr, tcpOptions, common.DefaultRootUsername, common.DefaultRootPassword) + if err != nil { + log.Fatalf("Connect as the root user: %v", err) + } + + tenants := make([]tenant, 0, tenantsCount) + for index := 1; index <= tenantsCount; index++ { + current := tenant{ + id: index, + stream: fmt.Sprintf(tenantStream, index), + user: fmt.Sprintf(producerUsername, index), + } + stream, err := ensureStream(ctx, root, current.stream) + if err != nil { + log.Fatalf("Prepare stream for tenant %d: %v", current.id, err) + } + if err := ensureTopic(ctx, root, stream.Id, topicName); err != nil { + log.Fatalf("Prepare topic for tenant %d: %v", current.id, err) + } + if err := ensureUser(ctx, root, current.user, producerPermissions(stream.Id)); err != nil { + log.Fatalf("Prepare user %q: %v", current.user, err) + } + tenants = append(tenants, current) + } + closeClient(root) + + for _, current := range tenants { + producer, err := dial(ctx, serverAddr, tcpOptions, current.user, tenantPassword) + if err != nil { + log.Fatalf("Connect as %q: %v", current.user, err) + } + if err := sendMessages(ctx, producer, current); err != nil { + closeClient(producer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + if err := ensureNoAccess(ctx, producer, current, tenants); err != nil { + closeClient(producer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + closeClient(producer) + } + + log.Printf("Multi-tenant producer finished, %d tenant(s) sent %d message(s) each to their own stream", + len(tenants), messagesPerBatch*batchesPerTenant) +} + +// sendMessages writes to the tenant's own stream with the tenant's own client. +func sendMessages(ctx context.Context, cli iggcon.Client, current tenant) error { + streamIdentifier, err := iggcon.NewIdentifier(current.stream) + if err != nil { + return err + } + topicIdentifier, err := iggcon.NewIdentifier(topicName) + if err != nil { + return err + } + + for batch := 1; batch <= batchesPerTenant; batch++ { + messages := make([]iggcon.IggyMessage, 0, messagesPerBatch) + for index := 1; index <= messagesPerBatch; index++ { + payload := fmt.Sprintf("event-%d-%d-tenant-%d", batch, index, current.id) + message, err := iggcon.NewIggyMessage([]byte(payload)) + if err != nil { + return fmt.Errorf("build message: %w", err) + } + messages = append(messages, message) + } + if _, err := cli.SendMessages(ctx, streamIdentifier, topicIdentifier, iggcon.PartitionId(partitionID), messages); err != nil { + return fmt.Errorf("send batch %d to stream %q: %w", batch, current.stream, err) + } + log.Printf("Tenant %d (%s) sent %d message(s) to %s -> %s", + current.id, current.user, messagesPerBatch, current.stream, topicName) + } + return nil +} + +// ensureNoAccess proves the tenant can read its own stream and cannot read any +// other tenant's stream. +func ensureNoAccess(ctx context.Context, cli iggcon.Client, current tenant, tenants []tenant) error { + own, err := iggcon.NewIdentifier(current.stream) + if err != nil { + return err + } + if _, err := cli.GetStream(ctx, own); err != nil { + return fmt.Errorf("tenant %s cannot read its own stream %q: %w", current.user, current.stream, err) + } + log.Printf("Tenant %d (%s) can read its own stream %q", current.id, current.user, current.stream) + + for _, other := range tenants { + if other.stream == current.stream { + continue + } + identifier, err := iggcon.NewIdentifier(other.stream) + if err != nil { + return err + } + if _, err := cli.GetStream(ctx, identifier); err != nil { + if errors.Is(err, ierror.ErrUnauthorized) || errors.Is(err, ierror.ErrStreamIdNotFound) { + log.Printf("Tenant %d (%s) was refused stream %q as expected", + current.id, current.user, other.stream) + continue + } + return fmt.Errorf("read stream %q as %s: %w", other.stream, current.user, err) + } + return fmt.Errorf("tenant %s can read stream %q, the server should have refused it", + current.user, other.stream) + } + return nil +} + +// ensureStream returns the named stream, creating it when this is the first run. +func ensureStream(ctx context.Context, cli iggcon.Client, name string) (*iggcon.StreamDetails, error) { + identifier, err := iggcon.NewIdentifier(name) + if err != nil { + return nil, err + } + stream, err := cli.GetStream(ctx, identifier) + if err == nil { + log.Printf("Reusing stream %q (id %d) from an earlier run", name, stream.Id) + return stream, nil + } + + stream, err = cli.CreateStream(ctx, name, nil) + if err == nil { + log.Printf("Created stream %q (id %d)", name, stream.Id) + return stream, nil + } + if !errors.Is(err, ierror.ErrStreamNameAlreadyExists) { + return nil, fmt.Errorf("create stream %q: %w", name, err) + } + stream, err = cli.GetStream(ctx, identifier) + if err != nil { + return nil, fmt.Errorf("read stream %q after it already existed: %w", name, err) + } + log.Printf("Reusing stream %q (id %d) from an earlier run", name, stream.Id) + return stream, nil +} + +// ensureTopic returns the named topic of the stream, creating it when missing. +func ensureTopic(ctx context.Context, cli iggcon.Client, streamID uint32, name string) error { + streamIdentifier, err := iggcon.NewIdentifier(streamID) + if err != nil { + return err + } + topicIdentifier, err := iggcon.NewIdentifier(name) + if err != nil { + return err + } + if _, err := cli.GetTopic(ctx, streamIdentifier, topicIdentifier); err == nil { + log.Printf("Reusing topic %q of stream %d", name, streamID) + return nil + } + + partitions := partitionsCount + if _, err := cli.CreateTopic(ctx, streamIdentifier, name, &iggcon.TopicOptions{PartitionsCount: &partitions}); err != nil { + if errors.Is(err, ierror.ErrTopicNameAlreadyExists) { + log.Printf("Reusing topic %q of stream %d", name, streamID) + return nil + } + return fmt.Errorf("create topic %q in stream %d: %w", name, streamID, err) + } + log.Printf("Created topic %q in stream %d", name, streamID) + return nil +} + +// ensureUser creates the user with the given permissions, or replaces the +// permissions of a user an earlier run left behind, so a second run is safe. +func ensureUser(ctx context.Context, cli iggcon.Client, username string, permissions *iggcon.Permissions) error { + identifier, err := iggcon.NewIdentifier(username) + if err != nil { + return err + } + if _, err := cli.GetUser(ctx, identifier); err == nil { Review Comment: simplification: `ensureUser` repeats the permission update after both existing-user paths, and the consumer repeats the same shape. put creation inside the failed lookup branch, then share the permission update and log. ########## examples/go/multi-tenant/consumer/main.go: ########## @@ -0,0 +1,325 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package main reads every tenant's stream as that tenant, with the tenant's +// own restricted credentials. +// +// The streams and their users come from the multi-tenant producer, so run it +// first. The consumer only adds its own read-only user per tenant, then proves +// that each of those users can read its own stream and no other. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "strconv" + "time" + + "github.com/apache/iggy/examples/go/common" + "github.com/apache/iggy/foreign/go/client" + "github.com/apache/iggy/foreign/go/client/tcp" + iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" +) + +const ( + tenantPassword = "secret" + topicName = "events" + partitionID = uint32(0) + messagesPerBatch = 5 + defaultTenants = 3 + consumerUsername = "tenant_%d_consumer" + tenantStream = "tenant_%d_stream" + connectTimeout = 30 * time.Second + pollAttempts = 20 + pollRetryInterval = 100 * time.Millisecond +) + +type tenant struct { + id int + stream string + user string +} + +func main() { + serverAddr, tcpOptions := getTcpOptions() + tenantsCount := tenantsFromEnv() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + log.Printf("Multi-tenant consumer started, tenants: %d", tenantsCount) + + root, err := dial(ctx, serverAddr, tcpOptions, common.DefaultRootUsername, common.DefaultRootPassword) + if err != nil { + log.Fatalf("Connect as the root user: %v", err) + } + + tenants := make([]tenant, 0, tenantsCount) + for index := 1; index <= tenantsCount; index++ { + current := tenant{ + id: index, + stream: fmt.Sprintf(tenantStream, index), + user: fmt.Sprintf(consumerUsername, index), + } + stream, err := existingStream(ctx, root, current.stream) + if err != nil { + closeClient(root) + log.Fatalf("Tenant %d: %v", current.id, err) + } + topic, err := existingTopic(ctx, root, stream.Id, topicName) + if err != nil { + closeClient(root) + log.Fatalf("Tenant %d: %v", current.id, err) + } + if err := ensureUser(ctx, root, current.user, consumerPermissions(stream.Id, topic.Id)); err != nil { + closeClient(root) + log.Fatalf("Prepare user %q: %v", current.user, err) + } + tenants = append(tenants, current) + } + closeClient(root) + + for _, current := range tenants { + consumer, err := dial(ctx, serverAddr, tcpOptions, current.user, tenantPassword) + if err != nil { + log.Fatalf("Connect as %q: %v", current.user, err) + } + received, err := receiveMessages(ctx, consumer, current) + if err != nil { + closeClient(consumer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + if err := ensureNoAccess(ctx, consumer, current, tenants); err != nil { + closeClient(consumer) + log.Fatalf("Tenant %d: %v", current.id, err) + } + log.Printf("Tenant %d (%s) read %d message(s) from its own stream %q", + current.id, current.user, received, current.stream) + closeClient(consumer) + } + + log.Printf("Multi-tenant consumer finished, %d tenant(s) read their own stream only", len(tenants)) +} + +// receiveMessages polls the tenant's own topic from the first offset, with the +// tenant's own client. +func receiveMessages(ctx context.Context, cli iggcon.Client, current tenant) (int, error) { + streamIdentifier, err := iggcon.NewIdentifier(current.stream) + if err != nil { + return 0, err + } + topicIdentifier, err := iggcon.NewIdentifier(topicName) + if err != nil { + return 0, err + } + + options := iggcon.DefaultPollOptions() + options.Count = messagesPerBatch + + for attempt := 1; ; attempt++ { + polled, err := cli.PollMessages(ctx, streamIdentifier, topicIdentifier, partitionID, + iggcon.DefaultConsumer(), iggcon.OffsetPollingStrategy(0), options) + if err != nil { + return 0, fmt.Errorf("poll stream %q: %w", current.stream, err) + } + if len(polled.Messages) == 0 { + if attempt >= pollAttempts { + return 0, fmt.Errorf("no messages in stream %q after %d polls: run the multi-tenant producer first", + current.stream, pollAttempts) + } + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-time.After(pollRetryInterval): + } + continue + } + for _, message := range polled.Messages { + log.Printf("Tenant %d (%s) received %q from %s at offset %d", + current.id, current.user, string(message.Payload), current.stream, message.Header.Offset) + } + return len(polled.Messages), nil + } +} + +// ensureNoAccess proves the tenant can read its own stream and cannot read any +// other tenant's stream. +func ensureNoAccess(ctx context.Context, cli iggcon.Client, current tenant, tenants []tenant) error { + own, err := iggcon.NewIdentifier(current.stream) + if err != nil { + return err + } + if _, err := cli.GetStream(ctx, own); err != nil { + return fmt.Errorf("tenant %s cannot read its own stream %q: %w", current.user, current.stream, err) + } + log.Printf("Tenant %d (%s) can read its own stream %q", current.id, current.user, current.stream) + + for _, other := range tenants { + if other.stream == current.stream { + continue + } + identifier, err := iggcon.NewIdentifier(other.stream) + if err != nil { + return err + } + if _, err := cli.GetStream(ctx, identifier); err != nil { + if errors.Is(err, ierror.ErrUnauthorized) || errors.Is(err, ierror.ErrStreamIdNotFound) { + log.Printf("Tenant %d (%s) was refused stream %q as expected", + current.id, current.user, other.stream) + continue + } + return fmt.Errorf("read stream %q as %s: %w", other.stream, current.user, err) + } + return fmt.Errorf("tenant %s can read stream %q, the server should have refused it", + current.user, other.stream) + } + return nil +} + +// existingStream returns a stream the producer created, and explains how to get +// one when it is missing. +func existingStream(ctx context.Context, cli iggcon.Client, name string) (*iggcon.StreamDetails, error) { + identifier, err := iggcon.NewIdentifier(name) + if err != nil { + return nil, err + } + stream, err := cli.GetStream(ctx, identifier) + if err != nil { + return nil, fmt.Errorf("stream %q is not available (%v): run the multi-tenant producer first", name, err) + } + return stream, nil +} + +// existingTopic returns a topic the producer created. +func existingTopic(ctx context.Context, cli iggcon.Client, streamID uint32, name string) (*iggcon.TopicDetails, error) { + streamIdentifier, err := iggcon.NewIdentifier(streamID) + if err != nil { + return nil, err + } + topicIdentifier, err := iggcon.NewIdentifier(name) + if err != nil { + return nil, err + } + topic, err := cli.GetTopic(ctx, streamIdentifier, topicIdentifier) + if err != nil { + return nil, fmt.Errorf("topic %q of stream %d is not available (%v): run the multi-tenant producer first", + name, streamID, err) + } + return topic, nil +} + +// ensureUser creates the user with the given permissions, or replaces the +// permissions of a user an earlier run left behind, so a second run is safe. +func ensureUser(ctx context.Context, cli iggcon.Client, username string, permissions *iggcon.Permissions) error { + identifier, err := iggcon.NewIdentifier(username) + if err != nil { + return err + } + if _, err := cli.GetUser(ctx, identifier); err == nil { + if err := cli.UpdatePermissions(ctx, identifier, permissions); err != nil { + return fmt.Errorf("update permissions of user %q: %w", username, err) + } + log.Printf("Reusing user %q with its permissions replaced", username) + return nil + } + + if _, err := cli.CreateUser(ctx, username, tenantPassword, iggcon.Active, permissions, nil); err == nil { + log.Printf("Created user %q", username) + return nil + } else if !errors.Is(err, ierror.ErrUserAlreadyExists) { + return fmt.Errorf("create user %q: %w", username, err) + } + if err := cli.UpdatePermissions(ctx, identifier, permissions); err != nil { + return fmt.Errorf("update permissions of user %q: %w", username, err) + } + log.Printf("Reusing user %q with its permissions replaced", username) + return nil +} + +// consumerPermissions grant reading the tenant's own stream and polling the +// one topic it carries. read_stream is what the server checks when a user asks +// for a stream, so a topic-only grant would hide the tenant's own stream. +func consumerPermissions(streamID, topicID uint32) *iggcon.Permissions { + return &iggcon.Permissions{ + Streams: map[int]*iggcon.StreamPermissions{ + int(streamID): { + ReadStream: true, + Topics: map[int]*iggcon.TopicPermissions{ Review Comment: simplification: `ReadStream` already grants the topic reads and message polls covered by this topic map. remove the nested grants and `topicID` parameter while retaining the topic existence check. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
