19676174607

Committed 25 Nov 2025 04:10PM UTC coverage: 41.687% (-0.06%) from 41.746%

Build # 19676174607

Build Type

push

github

Committed by

smartinov

Commit Message

chore: set default pprof address for http server

Run Details

0 of 3 new or added lines in 1 file covered. (0.0%)

875 of 2099 relevant lines covered (41.69%)

27538.76 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()
        // TODO: When keel is updated, set it in the correct place
        service.DefaultHTTPPProfAddr = ":6060"

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

foomo / contentserver / 19676174607

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