19674387362

Committed 25 Nov 2025 03:12PM UTC coverage: 41.746% (+0.08%) from 41.662%

Build # 19674387362

Build Type

Pull #67

github

Committed by

web-flow

Commit Message

Merge cabb48f16 into e7e5d09f5

Pull Request Pull Request #67: Add Multi-Cloud Blob Storage Support (AWS S3, Azure, GCS)

Run Details

166 of 366 new or added lines in 11 files covered. (45.36%)

3 existing lines in 2 files now uncovered.

875 of 2096 relevant lines covered (41.75%)

22387.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/cmd/http.go

package cmd

import (
        "context"
        "errors"
        "fmt"
        "strings"

        "github.com/foomo/contentserver/pkg/handler"
        "github.com/foomo/contentserver/pkg/repo"
        "github.com/foomo/keel"
        "github.com/foomo/keel/healthz"
        keelhttp "github.com/foomo/keel/net/http"
        "github.com/foomo/keel/net/http/middleware"
        "github.com/foomo/keel/service"
        "github.com/spf13/cobra"
        "github.com/spf13/viper"
        "go.uber.org/zap"
)

func NewHTTPCommand() *cobra.Command {
        v := newViper()
        cmd := &cobra.Command{
                Use:   "http <url>",
                Short: "Start http server",
                Args:  cobra.ExactArgs(1),
                ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
                        var comps []string
                        if len(args) == 0 {
                                comps = cobra.AppendActiveHelp(comps, "You must specify the URL for the repository you are adding")
                        } else {
                                comps = cobra.AppendActiveHelp(comps, "This command does not take any more arguments")
                        }
                        return comps, cobra.ShellCompDirectiveNoFileComp
                },
                RunE: func(cmd *cobra.Command, args []string) error {
                        svr := keel.NewServer(
                                keel.WithHTTPPrometheusService(servicePrometheusEnabledFlag(v)),
                                keel.WithHTTPHealthzService(serviceHealthzEnabledFlag(v)),
                                keel.WithPrometheusMeter(servicePrometheusEnabledFlag(v)),
                                keel.WithGracefulPeriod(gracefulPeriodFlag(v)),
                                keel.WithOTLPGRPCTracer(otelEnabledFlag(v)),
                                keel.WithHTTPPProfService(servicePProfEnabledFlag(v)),
                        )

                        l := svr.Logger()

                        // Create storage based on configuration
                        storage, err := createStorage(cmd.Context(), v, l)
                        if err != nil {
                                return fmt.Errorf("failed to create storage: %w", err)
                        }

                        history, err := repo.NewHistory(l.Named("inst.history"),
                                repo.HistoryWithStorage(storage),
                                repo.HistoryWithHistoryDir(historyDirFlag(v)),
                                repo.HistoryWithHistoryLimit(historyLimitFlag(v)),
                        )
                        if err != nil {
                                return fmt.Errorf("failed to create history: %w", err)
                        }

                        r := repo.New(l.Named("inst.repo"),
                                args[0],
                                history,
                                repo.WithHTTPClient(
                                        keelhttp.NewHTTPClient(
                                                keelhttp.HTTPClientWithTimeout(repositoryTimeoutFlag(v)),
                                                keelhttp.HTTPClientWithTelemetry(),
                                        ),
                                ),
                                repo.WithPollInterval(pollIntevalFlag(v)),
                                repo.WithPoll(pollFlag(v)),
                        )

                        isLoadedHealtherFn := healthz.NewHealthzerFn(func(ctx context.Context) error {
                                if !r.Loaded() {
                                        return errors.New("repo not loaded yet")
                                }
                                return nil
                        })
                        // start initial update and handle error
                        svr.AddStartupHealthzers(isLoadedHealtherFn)
                        svr.AddReadinessHealthzers(isLoadedHealtherFn)

                        svr.AddClosers(func(ctx context.Context) error {
                                return history.Close()
                        })

                        svr.AddServices(
                                service.NewGoRoutine(l.Named("go.repo"), "repo", func(ctx context.Context, l *zap.Logger) error {
                                        return r.Start(ctx)
                                }),
                                service.NewHTTP(l.Named("svc.http"), "http", addressFlag(v),
                                        handler.NewHTTP(l.Named("inst.handler"), r, handler.WithBasePath(basePathFlag(v))),
                                        middleware.Telemetry(),
                                        middleware.Logger(),
                                        middleware.GZip(),
                                        middleware.Recover(),
                                ),
                        )

                        svr.Run()
                        return nil
                },
        }

        flags := cmd.Flags()
        addAddressFlag(flags, v)
        addBasePathFlag(flags, v)
        addPollFlag(flags, v)
        addPollIntervalFlag(flags, v)
        addHistoryDirFlag(flags, v)
        addHistoryLimitFlag(flags, v)
        addShutdownTimeoutFlag(flags, v)
        addOtelEnabledFlag(flags, v)
        addServiceHealthzEnabledFlag(flags, v)
        addServicePrometheusEnabledFlag(flags, v)
        addServicePProfEnabledFlag(flags, v)
        addStorageTypeFlag(flags, v)
        addStorageBlobBucketFlag(flags, v)
        addStorageBlobPrefixFlag(flags, v)
        addRepositoryTimeoutFlag(flags, v)

        return cmd
}

// supportedBlobSchemes lists the URL schemes supported by blob storage
var supportedBlobSchemes = []string{"gs://", "s3://", "azblob://"}

// createStorage creates a storage backend based on the configuration
func createStorage(ctx context.Context, v *viper.Viper, l *zap.Logger) (repo.Storage, error) {
        storageType := storageTypeFlag(v)
        blobBucket := storageBlobBucketFlag(v)
        blobPrefix := storageBlobPrefixFlag(v)

        // Warn about ignored blob config
        if storageType != "blob" && (blobBucket != "" || blobPrefix != "") {
                l.Warn("blob storage flags are set but storage-type is not 'blob'; blob config will be ignored",
                        zap.String("storage-type", storageType),
                        zap.String("blob-bucket", blobBucket),
                        zap.String("blob-prefix", blobPrefix),
                )
        }

        l.Info("creating storage", zap.String("type", storageType))

        switch storageType {
        case "blob":
                if blobBucket == "" {
                        return nil, fmt.Errorf("blob bucket URL is required when storage-type is 'blob' (supported schemes: gs://, s3://, azblob://)")
                }
                if !isValidBlobScheme(blobBucket) {
                        return nil, fmt.Errorf("unsupported blob storage URL scheme in %q; supported schemes: gs://, s3://, azblob://", blobBucket)
                }
                l.Info("using blob storage",
                        zap.String("bucket", blobBucket),
                        zap.String("prefix", blobPrefix),
                        zap.String("provider", detectBlobProvider(blobBucket)),
                )
                return repo.NewBlobStorage(ctx, blobBucket, blobPrefix)
        case "filesystem", "":
                dir := historyDirFlag(v)
                l.Info("using filesystem storage", zap.String("dir", dir))
                return repo.NewFilesystemStorage(dir)
        default:
                return nil, fmt.Errorf("unknown storage type: %s (supported: filesystem, blob)", storageType)
        }
}

// isValidBlobScheme checks if the bucket URL has a supported scheme
func isValidBlobScheme(bucketURL string) bool {
        for _, scheme := range supportedBlobSchemes {
                if strings.HasPrefix(bucketURL, scheme) {
                        return true
                }
        }
        return false
}

// detectBlobProvider returns a human-readable provider name from the URL scheme
func detectBlobProvider(bucketURL string) string {
        switch {
        case strings.HasPrefix(bucketURL, "gs://"):
                return "Google Cloud Storage"
        case strings.HasPrefix(bucketURL, "s3://"):
                return "AWS S3"
        case strings.HasPrefix(bucketURL, "azblob://"):
                return "Azure Blob Storage"
        default:
                return "unknown"
        }
}

1	package cmd
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"strings"
8
9	"github.com/foomo/contentserver/pkg/handler"
10	"github.com/foomo/contentserver/pkg/repo"
11	"github.com/foomo/keel"
12	"github.com/foomo/keel/healthz"
13	keelhttp "github.com/foomo/keel/net/http"
14	"github.com/foomo/keel/net/http/middleware"
15	"github.com/foomo/keel/service"
16	"github.com/spf13/cobra"
17	"github.com/spf13/viper"
18	"go.uber.org/zap"
19	)
20
21	func NewHTTPCommand() *cobra.Command {	×
22	v := newViper()	×
23	cmd := &cobra.Command{	×
24	Use: "http <url>",	×
25	Short: "Start http server",	×
26	Args: cobra.ExactArgs(1),	×
27	ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {	×
28	var comps []string	×
29	if len(args) == 0 {	×
30	comps = cobra.AppendActiveHelp(comps, "You must specify the URL for the repository you are adding")	×
31	} else {	×
32	comps = cobra.AppendActiveHelp(comps, "This command does not take any more arguments")	×
33	}	×
34	return comps, cobra.ShellCompDirectiveNoFileComp	×
35	},
36	RunE: func(cmd *cobra.Command, args []string) error {	×
37	svr := keel.NewServer(	×
38	keel.WithHTTPPrometheusService(servicePrometheusEnabledFlag(v)),	×
39	keel.WithHTTPHealthzService(serviceHealthzEnabledFlag(v)),	×
40	keel.WithPrometheusMeter(servicePrometheusEnabledFlag(v)),	×
41	keel.WithGracefulPeriod(gracefulPeriodFlag(v)),	×
42	keel.WithOTLPGRPCTracer(otelEnabledFlag(v)),	×
NEW 43	keel.WithHTTPPProfService(servicePProfEnabledFlag(v)),	×
44	)	×
45		×
46	l := svr.Logger()	×
47		×
NEW 48	// Create storage based on configuration	×
NEW 49	storage, err := createStorage(cmd.Context(), v, l)	×
NEW 50	if err != nil {	×
NEW 51	return fmt.Errorf("failed to create storage: %w", err)	×
NEW 52	}	×
53
NEW 54	history, err := repo.NewHistory(l.Named("inst.history"),	×
NEW 55	repo.HistoryWithStorage(storage),	×
NEW 56	repo.HistoryWithHistoryDir(historyDirFlag(v)),	×
NEW 57	repo.HistoryWithHistoryLimit(historyLimitFlag(v)),	×
NEW 58	)	×
NEW 59	if err != nil {	×
NEW 60	return fmt.Errorf("failed to create history: %w", err)	×
NEW 61	}	×
62
63	r := repo.New(l.Named("inst.repo"),	×
64	args[0],	×
NEW 65	history,	×
66	repo.WithHTTPClient(	×
67	keelhttp.NewHTTPClient(	×
NEW 68	keelhttp.HTTPClientWithTimeout(repositoryTimeoutFlag(v)),	×
69	keelhttp.HTTPClientWithTelemetry(),	×
70	),	×
71	),	×
72	repo.WithPollInterval(pollIntevalFlag(v)),	×
73	repo.WithPoll(pollFlag(v)),	×
74	)	×
75		×
76	isLoadedHealtherFn := healthz.NewHealthzerFn(func(ctx context.Context) error {	×
77	if !r.Loaded() {	×
78	return errors.New("repo not loaded yet")	×
79	}	×
80	return nil	×
81	})
82	// start initial update and handle error
83	svr.AddStartupHealthzers(isLoadedHealtherFn)	×
84	svr.AddReadinessHealthzers(isLoadedHealtherFn)	×
85		×
NEW 86	svr.AddClosers(func(ctx context.Context) error {	×
NEW 87	return history.Close()	×
NEW 88	})	×
89
90	svr.AddServices(	×
91	service.NewGoRoutine(l.Named("go.repo"), "repo", func(ctx context.Context, l *zap.Logger) error {	×
92	return r.Start(ctx)	×
93	}),	×
94	service.NewHTTP(l.Named("svc.http"), "http", addressFlag(v),
95	handler.NewHTTP(l.Named("inst.handler"), r, handler.WithBasePath(basePathFlag(v))),
96	middleware.Telemetry(),
97	middleware.Logger(),
98	middleware.GZip(),
99	middleware.Recover(),
100	),
101	)
102
103	svr.Run()	×
104	return nil	×
105	},
106	}
107
108	flags := cmd.Flags()	×
109	addAddressFlag(flags, v)	×
110	addBasePathFlag(flags, v)	×
111	addPollFlag(flags, v)	×
112	addPollIntervalFlag(flags, v)	×
113	addHistoryDirFlag(flags, v)	×
114	addHistoryLimitFlag(flags, v)	×
115	addShutdownTimeoutFlag(flags, v)	×
116	addOtelEnabledFlag(flags, v)	×
117	addServiceHealthzEnabledFlag(flags, v)	×
118	addServicePrometheusEnabledFlag(flags, v)	×
NEW 119	addServicePProfEnabledFlag(flags, v)	×
NEW 120	addStorageTypeFlag(flags, v)	×
NEW 121	addStorageBlobBucketFlag(flags, v)	×
NEW 122	addStorageBlobPrefixFlag(flags, v)	×
NEW 123	addRepositoryTimeoutFlag(flags, v)	×
124		×
125	return cmd	×
126	}
127
128	// supportedBlobSchemes lists the URL schemes supported by blob storage
129	var supportedBlobSchemes = []string{"gs://", "s3://", "azblob://"}
130
131	// createStorage creates a storage backend based on the configuration
NEW 132	func createStorage(ctx context.Context, v viper.Viper, l zap.Logger) (repo.Storage, error) {	×
NEW 133	storageType := storageTypeFlag(v)	×
NEW 134	blobBucket := storageBlobBucketFlag(v)	×
NEW 135	blobPrefix := storageBlobPrefixFlag(v)	×
NEW 136		×
NEW 137	// Warn about ignored blob config	×
NEW 138	if storageType != "blob" && (blobBucket != "" \|\| blobPrefix != "") {	×
NEW 139	l.Warn("blob storage flags are set but storage-type is not 'blob'; blob config will be ignored",	×
NEW 140	zap.String("storage-type", storageType),	×
NEW 141	zap.String("blob-bucket", blobBucket),	×
NEW 142	zap.String("blob-prefix", blobPrefix),	×
NEW 143	)	×
NEW 144	}	×
145
NEW 146	l.Info("creating storage", zap.String("type", storageType))	×
NEW 147		×
NEW 148	switch storageType {	×
NEW 149	case "blob":	×
NEW 150	if blobBucket == "" {	×
NEW 151	return nil, fmt.Errorf("blob bucket URL is required when storage-type is 'blob' (supported schemes: gs://, s3://, azblob://)")	×
NEW 152	}	×
NEW 153	if !isValidBlobScheme(blobBucket) {	×
NEW 154	return nil, fmt.Errorf("unsupported blob storage URL scheme in %q; supported schemes: gs://, s3://, azblob://", blobBucket)	×
NEW 155	}	×
NEW 156	l.Info("using blob storage",	×
NEW 157	zap.String("bucket", blobBucket),	×
NEW 158	zap.String("prefix", blobPrefix),	×
NEW 159	zap.String("provider", detectBlobProvider(blobBucket)),	×
NEW 160	)	×
NEW 161	return repo.NewBlobStorage(ctx, blobBucket, blobPrefix)	×
NEW 162	case "filesystem", "":	×
NEW 163	dir := historyDirFlag(v)	×
NEW 164	l.Info("using filesystem storage", zap.String("dir", dir))	×
NEW 165	return repo.NewFilesystemStorage(dir)	×
NEW 166	default:	×
NEW 167	return nil, fmt.Errorf("unknown storage type: %s (supported: filesystem, blob)", storageType)	×
168	}
169	}
170
171	// isValidBlobScheme checks if the bucket URL has a supported scheme
NEW 172	func isValidBlobScheme(bucketURL string) bool {	×
NEW 173	for _, scheme := range supportedBlobSchemes {	×
NEW 174	if strings.HasPrefix(bucketURL, scheme) {	×
NEW 175	return true	×
NEW 176	}	×
177	}
NEW 178	return false	×
179	}
180
181	// detectBlobProvider returns a human-readable provider name from the URL scheme
NEW 182	func detectBlobProvider(bucketURL string) string {	×
NEW 183	switch {	×
NEW 184	case strings.HasPrefix(bucketURL, "gs://"):	×
NEW 185	return "Google Cloud Storage"	×
NEW 186	case strings.HasPrefix(bucketURL, "s3://"):	×
NEW 187	return "AWS S3"	×
NEW 188	case strings.HasPrefix(bucketURL, "azblob://"):	×
NEW 189	return "Azure Blob Storage"	×
NEW 190	default:	×
NEW 191	return "unknown"	×
192	}
193	}

foomo / contentserver / 19674387362

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous