serve.go

v1.3.7

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/cobra"
	"golang.org/x/mod/sumdb"

	"go.bigb.es/curator/internal/admin"
	"go.bigb.es/curator/internal/config"
	"go.bigb.es/curator/internal/dochtml"
	"go.bigb.es/curator/internal/git"
	"go.bigb.es/curator/internal/metrics"
	"go.bigb.es/curator/internal/oidcauth"
	"go.bigb.es/curator/internal/server"
	"go.bigb.es/curator/internal/storage"
	"go.bigb.es/curator/internal/store"
	isumdb "go.bigb.es/curator/internal/sumdb"
)

func serveCmd() *cobra.Command {
	var configPath string

	cmd := &cobra.Command{
		Use:   "serve",
		Short: "Start the curator HTTP server",
		RunE: func(cmd *cobra.Command, args []string) error {
			return runServe(configPath)
		},
	}

	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file (optional, env vars always apply)")

	return cmd
}

func runServe(configPath string) error {
	cfg, err := config.Load(configPath)
	if err != nil {
		return err
	}

	log.Printf("configuration:\n%s", cfg.Dump())

	// Open module store.
	moduleStore, err := store.Open(cfg.Database.Type, cfg.Database.DSN)
	if err != nil {
		return fmt.Errorf("open store: %w", err)
	}
	defer moduleStore.Close()

	docRenderer, err := dochtml.NewRenderer()
	if err != nil {
		return fmt.Errorf("init doc renderer: %w", err)
	}

	srv := &server.Server{
		Cfg:          cfg,
		Resolver:     moduleStore,
		Credentials:  moduleStore,
		ModuleLister: moduleStore,
		Git:          git.NewCache(cfg.Cache),
		HTTPClient:   &http.Client{Timeout: 30 * time.Second},
		DocRenderer:  docRenderer,
	}

	// Load auth tokens if configured.
	if cfg.AuthTokens != "" {
		tokens, err := server.ParseAuthTokens(cfg.AuthTokens)
		if err != nil {
			return fmt.Errorf("load auth tokens: %w", err)
		}

		srv.AuthTokens = tokens
		log.Printf("loaded %d auth tokens", len(tokens))
	}

	// Initialize storage if configured.
	if cfg.Storage != nil {
		st, err := initStorage(context.Background(), cfg.Storage)
		if err != nil {
			return fmt.Errorf("init storage: %w", err)
		}

		srv.Store = st
		storageType := cfg.Storage.Type
		if storageType == "" {
			storageType = "s3"
		}
		log.Printf("storage enabled: type=%s", storageType)
	}

	// Initialize sumdb if configured.
	var sumdbOps *isumdb.Ops

	if cfg.Sumdb != nil && cfg.Sumdb.Enabled {
		signer, err := config.ResolveValue(cfg.Sumdb.Key)
		if err != nil {
			return fmt.Errorf("read sumdb key: %w", err)
		}
		sumdbOps = isumdb.NewOps(cfg.Host, moduleStore, moduleStore, srv.Git, moduleStore, signer)
		log.Printf("sumdb enabled with %d records", sumdbOps.RecordCount())
	}

	// Metrics.
	metrics.Register()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var bucketStats metrics.BucketStatsFunc
	if srv.Store != nil {
		if sp, ok := srv.Store.(storage.StatsProvider); ok {
			bucketStats = sp.BucketStats
		}
	}

	var recordCount metrics.RecordCountFunc
	if sumdbOps != nil {
		recordCount = sumdbOps.RecordCount
	}

	moduleCount := func() int { return moduleStore.ModuleCount() }
	metrics.StartCollector(ctx, cfg.Cache, moduleCount, bucketStats, recordCount, 30*time.Second)

	// Instrumentation server (metrics + health) on a separate port.
	if inst := cfg.Instrumentation; inst != nil && inst.Listen != "" {
		instrMux := http.NewServeMux()

		metricsPath := inst.MetricsPath
		if metricsPath == "" {
			metricsPath = "/metrics"
		}
		instrMux.Handle(metricsPath, promhttp.Handler())

		healthPath := inst.HealthPath
		if healthPath == "" {
			healthPath = "/healthz"
		}
		instrMux.HandleFunc(healthPath, healthHandler)

		go func() {
			log.Printf("instrumentation serving on %s (metrics: %s, health: %s)",
				inst.Listen, metricsPath, healthPath)
			if err := http.ListenAndServe(inst.Listen, instrMux); err != nil {
				log.Fatalf("instrumentation server: %v", err)
			}
		}()
	}

	// Main HTTP mux.
	mux := http.NewServeMux()
	mux.Handle("/-/static/", http.StripPrefix("/-/static/", http.FileServer(dochtml.StaticFS())))

	// Admin UI and API.
	adminHandler, err := admin.New(moduleStore, version)
	if err != nil {
		return fmt.Errorf("init admin: %w", err)
	}

	apiHandler := admin.NewAPI(moduleStore)

	adminToken := cfg.AdminToken
	if adminToken != "" {
		parsed, err := server.ParseAdminToken(adminToken)
		if err != nil {
			return fmt.Errorf("parse admin token: %w", err)
		}
		adminToken = parsed
	}

	// Initialize OIDC if configured.
	var oidcProvider *oidcauth.Provider
	if cfg.OIDC != nil && cfg.OIDC.Issuer != "" {
		secret, err := config.ResolveValue(cfg.OIDC.ClientSecret)
		if err != nil {
			return fmt.Errorf("resolve oidc client_secret: %w", err)
		}
		oidcProvider, err = oidcauth.New(context.Background(), cfg.OIDC, secret)
		if err != nil {
			return fmt.Errorf("init oidc: %w", err)
		}
		mux.HandleFunc("/-/oidc/login", oidcProvider.LoginHandler)
		mux.HandleFunc("/-/oidc/callback", oidcProvider.CallbackHandler)
		mux.HandleFunc("/-/oidc/logout", oidcProvider.LogoutHandler)
		log.Printf("OIDC enabled: issuer=%s client_id=%s", cfg.OIDC.Issuer, cfg.OIDC.ClientID)
	}

	mux.Handle("/-/admin/", admin.AuthMiddleware(adminToken, oidcProvider, adminHandler))
	mux.Handle("/-/api/", admin.AuthMiddleware(adminToken, oidcProvider, apiHandler))

	if sumdbOps != nil {
		sumdbSrv := sumdb.NewServer(sumdbOps)
		prefix := "/sumdb/" + cfg.Host

		var upstream string
		if cfg.Sumdb.Upstream != "" {
			upstream = cfg.Sumdb.Upstream
			log.Printf("sumdb upstream proxy: %s", upstream)
		}

		handler := srv.SumdbHandler(
			http.StripPrefix(prefix, sumdbSrv),
			upstream, prefix,
		)
		mux.Handle(prefix+"/", srv.SumdbAuthMiddleware(handler))
		log.Printf("sumdb serving at %s/", prefix)
	}

	mux.Handle("/", srv)

	log.Printf("curator serving on %s (host: %s, db: %s/%s, modules: %d)",
		cfg.Listen, cfg.Host, cfg.Database.Type, cfg.Database.DSN, moduleStore.ModuleCount())

	return http.ListenAndServe(cfg.Listen, mux)
}

func initStorage(ctx context.Context, cfg *config.StorageConfig) (storage.Storage, error) {
	switch cfg.Type {
	case "s3", "":
		return storage.NewS3Storage(ctx, cfg.S3())
	case "disk":
		return storage.NewDiskStorage(cfg.Root)
	case "memory":
		return storage.NewMemoryStorage(cfg.MaxBytes), nil
	default:
		return nil, fmt.Errorf("unknown storage type: %q", cfg.Type)
	}
}

func healthHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.Write([]byte(`{"status":"ok"}` + "\n"))
}


1	package main
2
3	import (
4	"context"
5	"fmt"
6	"log"
7	"net/http"
8	"time"
9
10	"github.com/prometheus/client_golang/prometheus/promhttp"
11	"github.com/spf13/cobra"
12	"golang.org/x/mod/sumdb"
13
14	"go.bigb.es/curator/internal/admin"
15	"go.bigb.es/curator/internal/config"
16	"go.bigb.es/curator/internal/dochtml"
17	"go.bigb.es/curator/internal/git"
18	"go.bigb.es/curator/internal/metrics"
19	"go.bigb.es/curator/internal/oidcauth"
20	"go.bigb.es/curator/internal/server"
21	"go.bigb.es/curator/internal/storage"
22	"go.bigb.es/curator/internal/store"
23	isumdb "go.bigb.es/curator/internal/sumdb"
24	)
25
26	func serveCmd() *cobra.Command {
27	var configPath string
28
29	cmd := &cobra.Command{
30	Use: "serve",
31	Short: "Start the curator HTTP server",
32	RunE: func(cmd *cobra.Command, args []string) error {
33	return runServe(configPath)
34	},
35	}
36
37	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file (optional, env vars always apply)")
38
39	return cmd
40	}
41
42	func runServe(configPath string) error {
43	cfg, err := config.Load(configPath)
44	if err != nil {
45	return err
46	}
47
48	log.Printf("configuration:\n%s", cfg.Dump())
49
50	// Open module store.
51	moduleStore, err := store.Open(cfg.Database.Type, cfg.Database.DSN)
52	if err != nil {
53	return fmt.Errorf("open store: %w", err)
54	}
55	defer moduleStore.Close()
56
57	docRenderer, err := dochtml.NewRenderer()
58	if err != nil {
59	return fmt.Errorf("init doc renderer: %w", err)
60	}
61
62	srv := &server.Server{
63	Cfg: cfg,
64	Resolver: moduleStore,
65	Credentials: moduleStore,
66	ModuleLister: moduleStore,
67	Git: git.NewCache(cfg.Cache),
68	HTTPClient: &http.Client{Timeout: 30 * time.Second},
69	DocRenderer: docRenderer,
70	}
71
72	// Load auth tokens if configured.
73	if cfg.AuthTokens != "" {
74	tokens, err := server.ParseAuthTokens(cfg.AuthTokens)
75	if err != nil {
76	return fmt.Errorf("load auth tokens: %w", err)
77	}
78
79	srv.AuthTokens = tokens
80	log.Printf("loaded %d auth tokens", len(tokens))
81	}
82
83	// Initialize storage if configured.
84	if cfg.Storage != nil {
85	st, err := initStorage(context.Background(), cfg.Storage)
86	if err != nil {
87	return fmt.Errorf("init storage: %w", err)
88	}
89
90	srv.Store = st
91	storageType := cfg.Storage.Type
92	if storageType == "" {
93	storageType = "s3"
94	}
95	log.Printf("storage enabled: type=%s", storageType)
96	}
97
98	// Initialize sumdb if configured.
99	var sumdbOps *isumdb.Ops
100
101	if cfg.Sumdb != nil && cfg.Sumdb.Enabled {
102	signer, err := config.ResolveValue(cfg.Sumdb.Key)
103	if err != nil {
104	return fmt.Errorf("read sumdb key: %w", err)
105	}
106	sumdbOps = isumdb.NewOps(cfg.Host, moduleStore, moduleStore, srv.Git, moduleStore, signer)
107	log.Printf("sumdb enabled with %d records", sumdbOps.RecordCount())
108	}
109
110	// Metrics.
111	metrics.Register()
112
113	ctx, cancel := context.WithCancel(context.Background())
114	defer cancel()
115
116	var bucketStats metrics.BucketStatsFunc
117	if srv.Store != nil {
118	if sp, ok := srv.Store.(storage.StatsProvider); ok {
119	bucketStats = sp.BucketStats
120	}
121	}
122
123	var recordCount metrics.RecordCountFunc
124	if sumdbOps != nil {
125	recordCount = sumdbOps.RecordCount
126	}
127
128	moduleCount := func() int { return moduleStore.ModuleCount() }
129	metrics.StartCollector(ctx, cfg.Cache, moduleCount, bucketStats, recordCount, 30*time.Second)
130
131	// Instrumentation server (metrics + health) on a separate port.
132	if inst := cfg.Instrumentation; inst != nil && inst.Listen != "" {
133	instrMux := http.NewServeMux()
134
135	metricsPath := inst.MetricsPath
136	if metricsPath == "" {
137	metricsPath = "/metrics"
138	}
139	instrMux.Handle(metricsPath, promhttp.Handler())
140
141	healthPath := inst.HealthPath
142	if healthPath == "" {
143	healthPath = "/healthz"
144	}
145	instrMux.HandleFunc(healthPath, healthHandler)
146
147	go func() {
148	log.Printf("instrumentation serving on %s (metrics: %s, health: %s)",
149	inst.Listen, metricsPath, healthPath)
150	if err := http.ListenAndServe(inst.Listen, instrMux); err != nil {
151	log.Fatalf("instrumentation server: %v", err)
152	}
153	}()
154	}
155
156	// Main HTTP mux.
157	mux := http.NewServeMux()
158	mux.Handle("/-/static/", http.StripPrefix("/-/static/", http.FileServer(dochtml.StaticFS())))
159
160	// Admin UI and API.
161	adminHandler, err := admin.New(moduleStore, version)
162	if err != nil {
163	return fmt.Errorf("init admin: %w", err)
164	}
165
166	apiHandler := admin.NewAPI(moduleStore)
167
168	adminToken := cfg.AdminToken
169	if adminToken != "" {
170	parsed, err := server.ParseAdminToken(adminToken)
171	if err != nil {
172	return fmt.Errorf("parse admin token: %w", err)
173	}
174	adminToken = parsed
175	}
176
177	// Initialize OIDC if configured.
178	var oidcProvider *oidcauth.Provider
179	if cfg.OIDC != nil && cfg.OIDC.Issuer != "" {
180	secret, err := config.ResolveValue(cfg.OIDC.ClientSecret)
181	if err != nil {
182	return fmt.Errorf("resolve oidc client_secret: %w", err)
183	}
184	oidcProvider, err = oidcauth.New(context.Background(), cfg.OIDC, secret)
185	if err != nil {
186	return fmt.Errorf("init oidc: %w", err)
187	}
188	mux.HandleFunc("/-/oidc/login", oidcProvider.LoginHandler)
189	mux.HandleFunc("/-/oidc/callback", oidcProvider.CallbackHandler)
190	mux.HandleFunc("/-/oidc/logout", oidcProvider.LogoutHandler)
191	log.Printf("OIDC enabled: issuer=%s client_id=%s", cfg.OIDC.Issuer, cfg.OIDC.ClientID)
192	}
193
194	mux.Handle("/-/admin/", admin.AuthMiddleware(adminToken, oidcProvider, adminHandler))
195	mux.Handle("/-/api/", admin.AuthMiddleware(adminToken, oidcProvider, apiHandler))
196
197	if sumdbOps != nil {
198	sumdbSrv := sumdb.NewServer(sumdbOps)
199	prefix := "/sumdb/" + cfg.Host
200
201	var upstream string
202	if cfg.Sumdb.Upstream != "" {
203	upstream = cfg.Sumdb.Upstream
204	log.Printf("sumdb upstream proxy: %s", upstream)
205	}
206
207	handler := srv.SumdbHandler(
208	http.StripPrefix(prefix, sumdbSrv),
209	upstream, prefix,
210	)
211	mux.Handle(prefix+"/", srv.SumdbAuthMiddleware(handler))
212	log.Printf("sumdb serving at %s/", prefix)
213	}
214
215	mux.Handle("/", srv)
216
217	log.Printf("curator serving on %s (host: %s, db: %s/%s, modules: %d)",
218	cfg.Listen, cfg.Host, cfg.Database.Type, cfg.Database.DSN, moduleStore.ModuleCount())
219
220	return http.ListenAndServe(cfg.Listen, mux)
221	}
222
223	func initStorage(ctx context.Context, cfg *config.StorageConfig) (storage.Storage, error) {
224	switch cfg.Type {
225	case "s3", "":
226	return storage.NewS3Storage(ctx, cfg.S3())
227	case "disk":
228	return storage.NewDiskStorage(cfg.Root)
229	case "memory":
230	return storage.NewMemoryStorage(cfg.MaxBytes), nil
231	default:
232	return nil, fmt.Errorf("unknown storage type: %q", cfg.Type)
233	}
234	}
235
236	func healthHandler(w http.ResponseWriter, r *http.Request) {
237	w.Header().Set("Content-Type", "application/json")
238	w.Write([]byte(`{"status":"ok"}` + "\n"))
239	}
240

serve.go

Source Files