19671331757

Committed 25 Nov 2025 01:33PM UTC coverage: 42.455% (+0.8%) from 41.662%

Build # 19671331757

Build Type

Pull #67

github

Committed by

web-flow

Commit Message

Merge 12430bd68 into e7e5d09f5

Pull Request Pull Request #67: Add GCS Storage Backend Support

Run Details

166 of 331 new or added lines in 11 files covered. (50.15%)

3 existing lines in 2 files now uncovered.

875 of 2061 relevant lines covered (42.46%)

26984.83 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"

        "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.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)
        addStorageGCSBucketFlag(flags, v)
        addStorageGCSPrefixFlag(flags, v)

        return cmd
}

// 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)
        gcsBucket := storageGCSBucketFlag(v)
        gcsPrefix := storageGCSPrefixFlag(v)

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

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

        switch storageType {
        case "gcs":
                if gcsBucket == "" {
                        return nil, fmt.Errorf("GCS bucket URL is required for gcs storage type")
                }
                l.Info("using GCS storage", zap.String("bucket", gcsBucket), zap.String("prefix", gcsPrefix))
                return repo.NewBlobStorage(ctx, gcsBucket, gcsPrefix)
        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", storageType)
        }
}

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

foomo / contentserver / 19671331757

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