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
50 systemAuthToken string
51 debugLogf func(string, ...interface{})
53 ManagementToken string
59 theConfig = DefaultConfig()
60 formatter = map[string]logrus.Formatter{
61 "text": &logrus.TextFormatter{
63 TimestampFormat: rfc3339NanoFixed,
65 "json": &logrus.JSONFormatter{
66 TimestampFormat: rfc3339NanoFixed,
69 log = logrus.StandardLogger()
72 const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
74 // DefaultConfig returns the default configuration.
75 func DefaultConfig() *Config {
80 RequireSignatures: true,
81 BlobSignatureTTL: arvados.Duration(14 * 24 * time.Hour),
82 TrashLifetime: arvados.Duration(14 * 24 * time.Hour),
83 TrashCheckInterval: arvados.Duration(24 * time.Hour),
88 // Start should be called exactly once: after setting all public
89 // fields, and before using the config.
90 func (cfg *Config) Start() error {
92 log.Level = logrus.DebugLevel
93 cfg.debugLogf = log.Printf
94 cfg.debugLogf("debugging enabled")
96 log.Level = logrus.InfoLevel
97 cfg.debugLogf = func(string, ...interface{}) {}
100 f := formatter[strings.ToLower(cfg.LogFormat)]
102 return fmt.Errorf(`unsupported log format %q (try "text" or "json")`, cfg.LogFormat)
106 if cfg.MaxBuffers < 0 {
107 return fmt.Errorf("MaxBuffers must be greater than zero")
109 bufs = newBufferPool(cfg.MaxBuffers, BlockSize)
111 if cfg.MaxRequests < 1 {
112 cfg.MaxRequests = cfg.MaxBuffers * 2
113 log.Printf("MaxRequests <1 or not specified; defaulting to MaxBuffers * 2 == %d", cfg.MaxRequests)
116 if cfg.BlobSigningKeyFile != "" {
117 buf, err := ioutil.ReadFile(cfg.BlobSigningKeyFile)
119 return fmt.Errorf("reading blob signing key file: %s", err)
121 cfg.blobSigningKey = bytes.TrimSpace(buf)
122 if len(cfg.blobSigningKey) == 0 {
123 return fmt.Errorf("blob signing key file %q is empty", cfg.BlobSigningKeyFile)
125 } else if cfg.RequireSignatures {
126 return fmt.Errorf("cannot enable RequireSignatures (-enforce-permissions) without a blob signing key")
128 log.Println("Running without a blob signing key. Block locators " +
129 "returned by this server will not be signed, and will be rejected " +
130 "by a server that enforces permissions.")
131 log.Println("To fix this, use the BlobSigningKeyFile config entry.")
134 if fn := cfg.SystemAuthTokenFile; fn != "" {
135 buf, err := ioutil.ReadFile(fn)
137 return fmt.Errorf("cannot read system auth token file %q: %s", fn, err)
139 cfg.systemAuthToken = strings.TrimSpace(string(buf))
142 if cfg.EnableDelete {
143 log.Print("Trash/delete features are enabled. WARNING: this has not " +
144 "been extensively tested. You should disable this unless you can afford to lose data.")
147 if len(cfg.Volumes) == 0 {
148 if (&unixVolumeAdder{cfg}).Discover() == 0 {
149 return fmt.Errorf("no volumes found")
152 for _, v := range cfg.Volumes {
153 if err := v.Start(); err != nil {
154 return fmt.Errorf("volume %s: %s", v, err)
156 log.Printf("Using volume %v (writable=%v)", v, v.Writable())
161 type metrics struct {
162 registry *prometheus.Registry
163 reqDuration *prometheus.SummaryVec
164 timeToStatus *prometheus.SummaryVec
165 exportProm http.Handler
168 func (*metrics) Levels() []logrus.Level {
169 return logrus.AllLevels
172 func (m *metrics) Fire(ent *logrus.Entry) error {
173 if tts, ok := ent.Data["timeToStatus"].(stats.Duration); !ok {
174 } else if method, ok := ent.Data["reqMethod"].(string); !ok {
175 } else if code, ok := ent.Data["respStatusCode"].(int); !ok {
177 m.timeToStatus.WithLabelValues(strconv.Itoa(code), strings.ToLower(method)).Observe(time.Duration(tts).Seconds())
182 func (m *metrics) setup() {
183 m.registry = prometheus.NewRegistry()
184 m.timeToStatus = prometheus.NewSummaryVec(prometheus.SummaryOpts{
185 Name: "time_to_status_seconds",
186 Help: "Summary of request TTFB.",
187 }, []string{"code", "method"})
188 m.reqDuration = prometheus.NewSummaryVec(prometheus.SummaryOpts{
189 Name: "request_duration_seconds",
190 Help: "Summary of request duration.",
191 }, []string{"code", "method"})
192 m.registry.MustRegister(m.timeToStatus)
193 m.registry.MustRegister(m.reqDuration)
194 m.exportProm = promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{
200 func (m *metrics) exportJSON(w http.ResponseWriter, req *http.Request) {
201 jm := jsonpb.Marshaler{Indent: " "}
202 mfs, _ := m.registry.Gather()
204 for i, mf := range mfs {
213 func (m *metrics) Instrument(next http.Handler) http.Handler {
214 return promhttp.InstrumentHandlerDuration(m.reqDuration, next)
217 // VolumeTypes is built up by init() funcs in the source files that
218 // define the volume types.
219 var VolumeTypes = []func() VolumeWithExamples{}
221 type VolumeList []Volume
223 // UnmarshalJSON -- given an array of objects -- deserializes each
224 // object as the volume type indicated by the object's Type field.
225 func (vl *VolumeList) UnmarshalJSON(data []byte) error {
226 typeMap := map[string]func() VolumeWithExamples{}
227 for _, factory := range VolumeTypes {
228 t := factory().Type()
229 if _, ok := typeMap[t]; ok {
230 log.Fatal("volume type %+q is claimed by multiple VolumeTypes")
235 var mapList []map[string]interface{}
236 err := json.Unmarshal(data, &mapList)
240 for _, mapIn := range mapList {
241 typeIn, ok := mapIn["Type"].(string)
243 return fmt.Errorf("invalid volume type %+v", mapIn["Type"])
245 factory, ok := typeMap[typeIn]
247 return fmt.Errorf("unsupported volume type %+q", typeIn)
249 data, err := json.Marshal(mapIn)
254 err = json.Unmarshal(data, vol)
258 *vl = append(*vl, vol)
263 // MarshalJSON adds a "Type" field to each volume corresponding to its
265 func (vl *VolumeList) MarshalJSON() ([]byte, error) {
267 for _, vs := range *vl {
268 j, err := json.Marshal(vs)
273 data = append(data, byte(','))
275 t, err := json.Marshal(vs.Type())
279 data = append(data, j[0])
280 data = append(data, []byte(`"Type":`)...)
281 data = append(data, t...)
282 data = append(data, byte(','))
283 data = append(data, j[1:]...)
285 return append(data, byte(']')), nil