Skip to content

[Tavern] Correct GCP Subscription Prefix #824

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
KCarretto merged 3 commits into from
Jan 7, 2025
Merged

[Tavern] Correct GCP Subscription Prefix #824

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
48d33d3
remove gcpPrefix as the low-level client prepends it
KCarretto Jan 7, 2025
494562e
oops
KCarretto Jan 7, 2025
da4f603
Merge branch 'main' into strip-gcp-prefix
KCarretto Jan 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions tavern/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log"
"log/slog"
"net/http"
"strings"
"time"
Expand All @@ -23,7 +24,7 @@ import (
)

// GlobalInstanceID uniquely identifies this instance for logging and naming purposes.
var GlobalInstanceID = namegen.NewComplex()
var GlobalInstanceID = namegen.New()

var (
// EnvEnableTestData if set will populate the database with test data.
Expand Down Expand Up @@ -165,16 +166,22 @@ func (cfg *Config) NewShellMuxes(ctx context.Context) (wsMux *stream.Mux, grpcMu
}
defer client.Close()

createGCPSubscription := func(ctx context.Context, subName EnvString, topic *gcppubsub.Topic) string {
name := fmt.Sprintf("%s--%s", strings.TrimPrefix(subName.String(), gcpPrefix), GlobalInstanceID)
createGCPSubscription := func(ctx context.Context, topic *gcppubsub.Topic) string {
name := fmt.Sprintf("%s-sub_%s", topic.ID(), GlobalInstanceID)

sub, err := client.CreateSubscription(ctx, name, gcppubsub.SubscriptionConfig{
Topic: topic,
AckDeadline: 10 * time.Second,
ExpirationPolicy: 24 * time.Hour, // Automatically delete unused subscriptions after 1 day
})
if err != nil {
panic(fmt.Errorf("failed to create gcppubsub subscription (topic=%q), to disable creation do not use the 'gcppubsub://' prefix for the environment variable %q: %v", topic.ID(), EnvPubSubSubscriptionShellInput.Key, err))
panic(fmt.Errorf(
"failed to create gcppubsub subscription (topic=%q,subscription_name=%q), to disable creation do not use the 'gcppubsub://' prefix for the environment variable %q: %v",
topic.ID(),
name,
EnvPubSubSubscriptionShellInput.Key,
err,
))
}
exists, err := sub.Exists(ctx)
if err != nil {
Expand All @@ -190,8 +197,10 @@ func (cfg *Config) NewShellMuxes(ctx context.Context) (wsMux *stream.Mux, grpcMu
shellOutputTopic := client.Topic(strings.TrimPrefix(topicShellOutput, gcpPrefix))

// Overwrite env var specification with newly created GCP PubSub Subscriptions
subShellInput = fmt.Sprintf("gcpubsub://%s", createGCPSubscription(ctx, EnvPubSubSubscriptionShellInput, shellInputTopic))
subShellOutput = fmt.Sprintf("gcpubsub://%s", createGCPSubscription(ctx, EnvPubSubSubscriptionShellOutput, shellOutputTopic))
subShellInput = fmt.Sprintf("gcpubsub://%s", createGCPSubscription(ctx, shellInputTopic))
slog.DebugContext(ctx, "created GCP PubSub subscription for shell input", "subscription_name", subShellInput)
subShellOutput = fmt.Sprintf("gcpubsub://%s", createGCPSubscription(ctx, shellOutputTopic))
slog.DebugContext(ctx, "created GCP PubSub subscription for shell output", "subscription_name", subShellOutput)
}

pubOutput, err := pubsub.OpenTopic(ctx, topicShellOutput)
Expand Down