1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
17 "git.curoverse.com/arvados.git/sdk/go/arvados"
18 "git.curoverse.com/arvados.git/sdk/go/stats"
19 "github.com/Sirupsen/logrus"
20 "github.com/golang/protobuf/jsonpb"
21 "github.com/prometheus/client_golang/prometheus"
22 "github.com/prometheus/client_golang/prometheus/promhttp"
36 BlobSignatureTTL arvados.Duration
37 BlobSigningKeyFile string
38 RequireSignatures bool
39 SystemAuthTokenFile string
41 TrashLifetime arvados.Duration
42 TrashCheckInterval arvados.Duration
47 systemAuthToken string
48 debugLogf func(string, ...interface{})
50 ManagementToken string
56 theConfig = DefaultConfig()
57 formatter = map[string]logrus.Formatter{
58 "text": &logrus.TextFormatter{
60 TimestampFormat: rfc3339NanoFixed,
62 "json": &logrus.JSONFormatter{
63 TimestampFormat: rfc3339NanoFixed,
66 log = logrus.StandardLogger()
69 const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
71 // DefaultConfig returns the default configuration.
72 func DefaultConfig() *Config {
77 RequireSignatures: true,
78 BlobSignatureTTL: arvados.Duration(14 * 24 * time.Hour),
79 TrashLifetime: arvados.Duration(14 * 24 * time.Hour),
80 TrashCheckInterval: arvados.Duration(24 * time.Hour),
85 // Start should be called exactly once: after setting all public
86 // fields, and before using the config.
87 func (cfg *Config) Start() error {
89 log.Level = logrus.DebugLevel
90 cfg.debugLogf = log.Printf
91 cfg.debugLogf("debugging enabled")
93 log.Level = logrus.InfoLevel
94 cfg.debugLogf = func(string, ...interface{}) {}
97 f := formatter[strings.ToLower(cfg.LogFormat)]
99 return fmt.Errorf(`unsupported log format %q (try "text" or "json")`, cfg.LogFormat)
103 if cfg.MaxBuffers < 0 {
104 return fmt.Errorf("MaxBuffers must be greater than zero")
106 bufs = newBufferPool(cfg.MaxBuffers, BlockSize)
108 if cfg.MaxRequests < 1 {
109 cfg.MaxRequests = cfg.MaxBuffers * 2
110 log.Printf("MaxRequests <1 or not specified; defaulting to MaxBuffers * 2 == %d", cfg.MaxRequests)
113 if cfg.BlobSigningKeyFile != "" {
114 buf, err := ioutil.ReadFile(cfg.BlobSigningKeyFile)
116 return fmt.Errorf("reading blob signing key file: %s", err)
118 cfg.blobSigningKey = bytes.TrimSpace(buf)
119 if len(cfg.blobSigningKey) == 0 {
120 return fmt.Errorf("blob signing key file %q is empty", cfg.BlobSigningKeyFile)
122 } else if cfg.RequireSignatures {
123 return fmt.Errorf("cannot enable RequireSignatures (-enforce-permissions) without a blob signing key")
125 log.Println("Running without a blob signing key. Block locators " +
126 "returned by this server will not be signed, and will be rejected " +
127 "by a server that enforces permissions.")
128 log.Println("To fix this, use the BlobSigningKeyFile config entry.")
131 if fn := cfg.SystemAuthTokenFile; fn != "" {
132 buf, err := ioutil.ReadFile(fn)
134 return fmt.Errorf("cannot read system auth token file %q: %s", fn, err)
136 cfg.systemAuthToken = strings.TrimSpace(string(buf))
139 if cfg.EnableDelete {
140 log.Print("Trash/delete features are enabled. WARNING: this has not " +
141 "been extensively tested. You should disable this unless you can afford to lose data.")
144 if len(cfg.Volumes) == 0 {
145 if (&unixVolumeAdder{cfg}).Discover() == 0 {
146 return fmt.Errorf("no volumes found")
149 for _, v := range cfg.Volumes {
150 if err := v.Start(); err != nil {
151 return fmt.Errorf("volume %s: %s", v, err)
153 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
158 type metrics struct {
159 registry *prometheus.Registry
160 reqDuration *prometheus.SummaryVec
161 timeToStatus *prometheus.SummaryVec
162 exportProm http.Handler
165 func (*metrics) Levels() []logrus.Level {
166 return logrus.AllLevels
169 func (m *metrics) Fire(ent *logrus.Entry) error {
170 if tts, ok := ent.Data["timeToStatus"].(stats.Duration); !ok {
171 } else if method, ok := ent.Data["reqMethod"].(string); !ok {
172 } else if code, ok := ent.Data["respStatusCode"].(int); !ok {
174 m.timeToStatus.WithLabelValues(strconv.Itoa(code), strings.ToLower(method)).Observe(time.Duration(tts).Seconds())
179 func (m *metrics) setup() {
180 m.registry = prometheus.NewRegistry()
181 m.timeToStatus = prometheus.NewSummaryVec(prometheus.SummaryOpts{
182 Name: "time_to_status_seconds",
183 Help: "Summary of request TTFB.",
184 }, []string{"code", "method"})
185 m.reqDuration = prometheus.NewSummaryVec(prometheus.SummaryOpts{
186 Name: "request_duration_seconds",
187 Help: "Summary of request duration.",
188 }, []string{"code", "method"})
189 m.registry.MustRegister(m.timeToStatus)
190 m.registry.MustRegister(m.reqDuration)
191 m.exportProm = promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{
197 func (m *metrics) exportJSON(w http.ResponseWriter, req *http.Request) {
198 jm := jsonpb.Marshaler{Indent: " "}
199 mfs, _ := m.registry.Gather()
201 for i, mf := range mfs {
210 func (m *metrics) Instrument(next http.Handler) http.Handler {
211 return promhttp.InstrumentHandlerDuration(m.reqDuration, next)
214 // VolumeTypes is built up by init() funcs in the source files that
215 // define the volume types.
216 var VolumeTypes = []func() VolumeWithExamples{}
218 type VolumeList []Volume
220 // UnmarshalJSON -- given an array of objects -- deserializes each
221 // object as the volume type indicated by the object's Type field.
222 func (vl *VolumeList) UnmarshalJSON(data []byte) error {
223 typeMap := map[string]func() VolumeWithExamples{}
224 for _, factory := range VolumeTypes {
225 t := factory().Type()
226 if _, ok := typeMap[t]; ok {
227 log.Fatal("volume type %+q is claimed by multiple VolumeTypes")
232 var mapList []map[string]interface{}
233 err := json.Unmarshal(data, &mapList)
237 for _, mapIn := range mapList {
238 typeIn, ok := mapIn["Type"].(string)
240 return fmt.Errorf("invalid volume type %+v", mapIn["Type"])
242 factory, ok := typeMap[typeIn]
244 return fmt.Errorf("unsupported volume type %+q", typeIn)
246 data, err := json.Marshal(mapIn)
251 err = json.Unmarshal(data, vol)
255 *vl = append(*vl, vol)
260 // MarshalJSON adds a "Type" field to each volume corresponding to its
262 func (vl *VolumeList) MarshalJSON() ([]byte, error) {
264 for _, vs := range *vl {
265 j, err := json.Marshal(vs)
270 data = append(data, byte(','))
272 t, err := json.Marshal(vs.Type())
276 data = append(data, j[0])
277 data = append(data, []byte(`"Type":`)...)
278 data = append(data, t...)
279 data = append(data, byte(','))
280 data = append(data, j[1:]...)
282 return append(data, byte(']')), nil