prometheus

1
// Copyright 2016 The Prometheus Authors
2
// Licensed under the Apache License, Version 2.0 (the "License");
3
// you may not use this file except in compliance with the License.
4
// You may obtain a copy of the License at
5
//
6
//     http://www.apache.org/licenses/LICENSE-2.0
7
//
8
// Unless required by applicable law or agreed to in writing, software
9
// distributed under the License is distributed on an "AS IS" BASIS,
10
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11
// See the License for the specific language governing permissions and
12
// limitations under the License.
13

14
package web
15

16
import (
17
	"context"
18
	"encoding/json"
19
	"fmt"
20
	"io"
21
	"net"
22
	"net/http"
23
	"net/http/httptest"
24
	"net/url"
25
	"os"
26
	"path/filepath"
27
	"strconv"
28
	"strings"
29
	"sync"
30
	"testing"
31
	"time"
32

33
	"github.com/prometheus/client_golang/prometheus"
34
	prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
35
	"github.com/stretchr/testify/require"
36

37
	"github.com/prometheus/prometheus/config"
38
	"github.com/prometheus/prometheus/notifier"
39
	"github.com/prometheus/prometheus/rules"
40
	"github.com/prometheus/prometheus/scrape"
41
	"github.com/prometheus/prometheus/tsdb"
42
	"github.com/prometheus/prometheus/util/testutil"
43
)
44

45
func TestMain(m *testing.M) {
46
	// On linux with a global proxy the tests will fail as the go client(http,grpc) tries to connect through the proxy.
47
	os.Setenv("no_proxy", "localhost,127.0.0.1,0.0.0.0,:")
48
	os.Exit(m.Run())
49
}
50

51
type dbAdapter struct {
52
	*tsdb.DB
53
}
54

55
func (a *dbAdapter) Stats(statsByLabelName string, limit int) (*tsdb.Stats, error) {
56
	return a.Head().Stats(statsByLabelName, limit), nil
57
}
58

59
func (a *dbAdapter) WALReplayStatus() (tsdb.WALReplayStatus, error) {
60
	return tsdb.WALReplayStatus{}, nil
61
}
62

63
func TestReadyAndHealthy(t *testing.T) {
64
	t.Parallel()
65

66
	dbDir := t.TempDir()
67

68
	db, err := tsdb.Open(dbDir, nil, nil, nil, nil)
69
	require.NoError(t, err)
70
	t.Cleanup(func() {
71
		require.NoError(t, db.Close())
72
	})
73
	port := fmt.Sprintf(":%d", testutil.RandomUnprivilegedPort(t))
74

75
	opts := &Options{
76
		ListenAddress:  port,
77
		ReadTimeout:    30 * time.Second,
78
		MaxConnections: 512,
79
		Context:        nil,
80
		Storage:        nil,
81
		LocalStorage:   &dbAdapter{db},
82
		TSDBDir:        dbDir,
83
		QueryEngine:    nil,
84
		ScrapeManager:  &scrape.Manager{},
85
		RuleManager:    &rules.Manager{},
86
		Notifier:       nil,
87
		RoutePrefix:    "/",
88
		EnableAdminAPI: true,
89
		ExternalURL: &url.URL{
90
			Scheme: "http",
91
			Host:   "localhost" + port,
92
			Path:   "/",
93
		},
94
		Version:  &PrometheusVersion{},
95
		Gatherer: prometheus.DefaultGatherer,
96
	}
97

98
	opts.Flags = map[string]string{}
99

100
	webHandler := New(nil, opts)
101

102
	webHandler.config = &config.Config{}
103
	webHandler.notifier = &notifier.Manager{}
104
	l, err := webHandler.Listener()
105
	if err != nil {
106
		panic(fmt.Sprintf("Unable to start web listener: %s", err))
107
	}
108

109
	ctx, cancel := context.WithCancel(context.Background())
110
	defer cancel()
111
	go func() {
112
		err := webHandler.Run(ctx, l, "")
113
		if err != nil {
114
			panic(fmt.Sprintf("Can't start web handler:%s", err))
115
		}
116
	}()
117

118
	// Give some time for the web goroutine to run since we need the server
119
	// to be up before starting tests.
120
	time.Sleep(5 * time.Second)
121

122
	baseURL := "http://localhost" + port
123

124
	resp, err := http.Get(baseURL + "/-/healthy")
125
	require.NoError(t, err)
126
	require.Equal(t, http.StatusOK, resp.StatusCode)
127
	cleanupTestResponse(t, resp)
128

129
	resp, err = http.Head(baseURL + "/-/healthy")
130
	require.NoError(t, err)
131
	require.Equal(t, http.StatusOK, resp.StatusCode)
132
	cleanupTestResponse(t, resp)
133

134
	for _, u := range []string{
135
		baseURL + "/-/ready",
136
	} {
137
		resp, err = http.Get(u)
138
		require.NoError(t, err)
139
		require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
140
		cleanupTestResponse(t, resp)
141

142
		resp, err = http.Head(u)
143
		require.NoError(t, err)
144
		require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
145
		cleanupTestResponse(t, resp)
146
	}
147

148
	resp, err = http.Post(baseURL+"/api/v1/admin/tsdb/snapshot", "", strings.NewReader(""))
149
	require.NoError(t, err)
150
	require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
151
	cleanupTestResponse(t, resp)
152

153
	resp, err = http.Post(baseURL+"/api/v1/admin/tsdb/delete_series", "", strings.NewReader("{}"))
154
	require.NoError(t, err)
155
	require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
156
	cleanupTestResponse(t, resp)
157

158
	// Set to ready.
159
	webHandler.SetReady(true)
160

161
	for _, u := range []string{
162
		baseURL + "/-/healthy",
163
		baseURL + "/-/ready",
164
	} {
165
		resp, err = http.Get(u)
166
		require.NoError(t, err)
167
		require.Equal(t, http.StatusOK, resp.StatusCode)
168
		cleanupTestResponse(t, resp)
169

170
		resp, err = http.Head(u)
171
		require.NoError(t, err)
172
		require.Equal(t, http.StatusOK, resp.StatusCode)
173
		cleanupTestResponse(t, resp)
174
	}
175

176
	resp, err = http.Post(baseURL+"/api/v1/admin/tsdb/snapshot", "", strings.NewReader(""))
177
	require.NoError(t, err)
178
	require.Equal(t, http.StatusOK, resp.StatusCode)
179
	cleanupSnapshot(t, dbDir, resp)
180
	cleanupTestResponse(t, resp)
181

182
	resp, err = http.Post(baseURL+"/api/v1/admin/tsdb/delete_series?match[]=up", "", nil)
183
	require.NoError(t, err)
184
	require.Equal(t, http.StatusNoContent, resp.StatusCode)
185
	cleanupTestResponse(t, resp)
186
}
187

188
func TestRoutePrefix(t *testing.T) {
189
	t.Parallel()
190
	dbDir := t.TempDir()
191

192
	db, err := tsdb.Open(dbDir, nil, nil, nil, nil)
193
	require.NoError(t, err)
194
	t.Cleanup(func() {
195
		require.NoError(t, db.Close())
196
	})
197

198
	port := fmt.Sprintf(":%d", testutil.RandomUnprivilegedPort(t))
199

200
	opts := &Options{
201
		ListenAddress:  port,
202
		ReadTimeout:    30 * time.Second,
203
		MaxConnections: 512,
204
		Context:        nil,
205
		TSDBDir:        dbDir,
206
		LocalStorage:   &dbAdapter{db},
207
		Storage:        nil,
208
		QueryEngine:    nil,
209
		ScrapeManager:  nil,
210
		RuleManager:    nil,
211
		Notifier:       nil,
212
		RoutePrefix:    "/prometheus",
213
		EnableAdminAPI: true,
214
		ExternalURL: &url.URL{
215
			Host:   "localhost.localdomain" + port,
216
			Scheme: "http",
217
		},
218
	}
219

220
	opts.Flags = map[string]string{}
221

222
	webHandler := New(nil, opts)
223
	l, err := webHandler.Listener()
224
	if err != nil {
225
		panic(fmt.Sprintf("Unable to start web listener: %s", err))
226
	}
227
	ctx, cancel := context.WithCancel(context.Background())
228
	defer cancel()
229
	go func() {
230
		err := webHandler.Run(ctx, l, "")
231
		if err != nil {
232
			panic(fmt.Sprintf("Can't start web handler:%s", err))
233
		}
234
	}()
235

236
	// Give some time for the web goroutine to run since we need the server
237
	// to be up before starting tests.
238
	time.Sleep(5 * time.Second)
239

240
	baseURL := "http://localhost" + port
241

242
	resp, err := http.Get(baseURL + opts.RoutePrefix + "/-/healthy")
243
	require.NoError(t, err)
244
	require.Equal(t, http.StatusOK, resp.StatusCode)
245
	cleanupTestResponse(t, resp)
246

247
	resp, err = http.Get(baseURL + opts.RoutePrefix + "/-/ready")
248
	require.NoError(t, err)
249
	require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
250
	cleanupTestResponse(t, resp)
251

252
	resp, err = http.Post(baseURL+opts.RoutePrefix+"/api/v1/admin/tsdb/snapshot", "", strings.NewReader(""))
253
	require.NoError(t, err)
254
	require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
255
	cleanupTestResponse(t, resp)
256

257
	resp, err = http.Post(baseURL+opts.RoutePrefix+"/api/v1/admin/tsdb/delete_series", "", strings.NewReader("{}"))
258
	require.NoError(t, err)
259
	require.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
260
	cleanupTestResponse(t, resp)
261

262
	// Set to ready.
263
	webHandler.SetReady(true)
264

265
	resp, err = http.Get(baseURL + opts.RoutePrefix + "/-/healthy")
266
	require.NoError(t, err)
267
	require.Equal(t, http.StatusOK, resp.StatusCode)
268
	cleanupTestResponse(t, resp)
269

270
	resp, err = http.Get(baseURL + opts.RoutePrefix + "/-/ready")
271
	require.NoError(t, err)
272
	require.Equal(t, http.StatusOK, resp.StatusCode)
273
	cleanupTestResponse(t, resp)
274

275
	resp, err = http.Post(baseURL+opts.RoutePrefix+"/api/v1/admin/tsdb/snapshot", "", strings.NewReader(""))
276
	require.NoError(t, err)
277
	require.Equal(t, http.StatusOK, resp.StatusCode)
278
	cleanupSnapshot(t, dbDir, resp)
279
	cleanupTestResponse(t, resp)
280

281
	resp, err = http.Post(baseURL+opts.RoutePrefix+"/api/v1/admin/tsdb/delete_series?match[]=up", "", nil)
282
	require.NoError(t, err)
283
	require.Equal(t, http.StatusNoContent, resp.StatusCode)
284
	cleanupTestResponse(t, resp)
285
}
286

287
func TestDebugHandler(t *testing.T) {
288
	for _, tc := range []struct {
289
		prefix, url string
290
		code        int
291
	}{
292
		{"/", "/debug/pprof/cmdline", 200},
293
		{"/foo", "/foo/debug/pprof/cmdline", 200},
294

295
		{"/", "/debug/pprof/goroutine", 200},
296
		{"/foo", "/foo/debug/pprof/goroutine", 200},
297

298
		{"/", "/debug/pprof/foo", 404},
299
		{"/foo", "/bar/debug/pprof/goroutine", 404},
300
	} {
301
		opts := &Options{
302
			RoutePrefix:   tc.prefix,
303
			ListenAddress: "somehost:9090",
304
			ExternalURL: &url.URL{
305
				Host:   "localhost.localdomain:9090",
306
				Scheme: "http",
307
			},
308
		}
309
		handler := New(nil, opts)
310
		handler.SetReady(true)
311

312
		w := httptest.NewRecorder()
313

314
		req, err := http.NewRequest(http.MethodGet, tc.url, nil)
315

316
		require.NoError(t, err)
317

318
		handler.router.ServeHTTP(w, req)
319

320
		require.Equal(t, tc.code, w.Code)
321
	}
322
}
323

324
func TestHTTPMetrics(t *testing.T) {
325
	t.Parallel()
326
	handler := New(nil, &Options{
327
		RoutePrefix:   "/",
328
		ListenAddress: "somehost:9090",
329
		ExternalURL: &url.URL{
330
			Host:   "localhost.localdomain:9090",
331
			Scheme: "http",
332
		},
333
	})
334
	getReady := func() int {
335
		t.Helper()
336
		w := httptest.NewRecorder()
337

338
		req, err := http.NewRequest(http.MethodGet, "/-/ready", nil)
339
		require.NoError(t, err)
340

341
		handler.router.ServeHTTP(w, req)
342
		return w.Code
343
	}
344

345
	code := getReady()
346
	require.Equal(t, http.StatusServiceUnavailable, code)
347
	ready := handler.metrics.readyStatus
348
	require.Equal(t, 0, int(prom_testutil.ToFloat64(ready)))
349
	counter := handler.metrics.requestCounter
350
	require.Equal(t, 1, int(prom_testutil.ToFloat64(counter.WithLabelValues("/-/ready", strconv.Itoa(http.StatusServiceUnavailable)))))
351

352
	handler.SetReady(true)
353
	for range [2]int{} {
354
		code = getReady()
355
		require.Equal(t, http.StatusOK, code)
356
	}
357
	require.Equal(t, 1, int(prom_testutil.ToFloat64(ready)))
358
	require.Equal(t, 2, int(prom_testutil.ToFloat64(counter.WithLabelValues("/-/ready", strconv.Itoa(http.StatusOK)))))
359
	require.Equal(t, 1, int(prom_testutil.ToFloat64(counter.WithLabelValues("/-/ready", strconv.Itoa(http.StatusServiceUnavailable)))))
360

361
	handler.SetReady(false)
362
	for range [2]int{} {
363
		code = getReady()
364
		require.Equal(t, http.StatusServiceUnavailable, code)
365
	}
366
	require.Equal(t, 0, int(prom_testutil.ToFloat64(ready)))
367
	require.Equal(t, 2, int(prom_testutil.ToFloat64(counter.WithLabelValues("/-/ready", strconv.Itoa(http.StatusOK)))))
368
	require.Equal(t, 3, int(prom_testutil.ToFloat64(counter.WithLabelValues("/-/ready", strconv.Itoa(http.StatusServiceUnavailable)))))
369
}
370

371
func TestShutdownWithStaleConnection(t *testing.T) {
372
	dbDir := t.TempDir()
373

374
	db, err := tsdb.Open(dbDir, nil, nil, nil, nil)
375
	require.NoError(t, err)
376
	t.Cleanup(func() {
377
		require.NoError(t, db.Close())
378
	})
379
	timeout := 10 * time.Second
380

381
	port := fmt.Sprintf(":%d", testutil.RandomUnprivilegedPort(t))
382

383
	opts := &Options{
384
		ListenAddress:  port,
385
		ReadTimeout:    timeout,
386
		MaxConnections: 512,
387
		Context:        nil,
388
		Storage:        nil,
389
		LocalStorage:   &dbAdapter{db},
390
		TSDBDir:        dbDir,
391
		QueryEngine:    nil,
392
		ScrapeManager:  &scrape.Manager{},
393
		RuleManager:    &rules.Manager{},
394
		Notifier:       nil,
395
		RoutePrefix:    "/",
396
		ExternalURL: &url.URL{
397
			Scheme: "http",
398
			Host:   "localhost" + port,
399
			Path:   "/",
400
		},
401
		Version:  &PrometheusVersion{},
402
		Gatherer: prometheus.DefaultGatherer,
403
	}
404

405
	opts.Flags = map[string]string{}
406

407
	webHandler := New(nil, opts)
408

409
	webHandler.config = &config.Config{}
410
	webHandler.notifier = &notifier.Manager{}
411
	l, err := webHandler.Listener()
412
	if err != nil {
413
		panic(fmt.Sprintf("Unable to start web listener: %s", err))
414
	}
415

416
	closed := make(chan struct{})
417

418
	ctx, cancel := context.WithCancel(context.Background())
419
	go func() {
420
		err := webHandler.Run(ctx, l, "")
421
		if err != nil {
422
			panic(fmt.Sprintf("Can't start web handler:%s", err))
423
		}
424
		close(closed)
425
	}()
426

427
	// Give some time for the web goroutine to run since we need the server
428
	// to be up before starting tests.
429
	time.Sleep(5 * time.Second)
430

431
	// Open a socket, and don't use it. This connection should then be closed
432
	// after the ReadTimeout.
433
	c, err := net.Dial("tcp", opts.ExternalURL.Host)
434
	require.NoError(t, err)
435
	t.Cleanup(func() { require.NoError(t, c.Close()) })
436

437
	// Stop the web handler.
438
	cancel()
439

440
	select {
441
	case <-closed:
442
	case <-time.After(timeout + 5*time.Second):
443
		require.FailNow(t, "Server still running after read timeout.")
444
	}
445
}
446

447
func TestHandleMultipleQuitRequests(t *testing.T) {
448
	port := fmt.Sprintf(":%d", testutil.RandomUnprivilegedPort(t))
449

450
	opts := &Options{
451
		ListenAddress:   port,
452
		MaxConnections:  512,
453
		EnableLifecycle: true,
454
		RoutePrefix:     "/",
455
		ExternalURL: &url.URL{
456
			Scheme: "http",
457
			Host:   "localhost" + port,
458
			Path:   "/",
459
		},
460
	}
461
	webHandler := New(nil, opts)
462
	webHandler.config = &config.Config{}
463
	webHandler.notifier = &notifier.Manager{}
464
	l, err := webHandler.Listener()
465
	if err != nil {
466
		panic(fmt.Sprintf("Unable to start web listener: %s", err))
467
	}
468
	ctx, cancel := context.WithCancel(context.Background())
469
	closed := make(chan struct{})
470
	go func() {
471
		err := webHandler.Run(ctx, l, "")
472
		if err != nil {
473
			panic(fmt.Sprintf("Can't start web handler:%s", err))
474
		}
475
		close(closed)
476
	}()
477

478
	// Give some time for the web goroutine to run since we need the server
479
	// to be up before starting tests.
480
	time.Sleep(5 * time.Second)
481

482
	baseURL := opts.ExternalURL.Scheme + "://" + opts.ExternalURL.Host
483

484
	start := make(chan struct{})
485
	var wg sync.WaitGroup
486
	for i := 0; i < 3; i++ {
487
		wg.Add(1)
488
		go func() {
489
			defer wg.Done()
490
			<-start
491
			resp, err := http.Post(baseURL+"/-/quit", "", strings.NewReader(""))
492
			require.NoError(t, err)
493
			require.Equal(t, http.StatusOK, resp.StatusCode)
494
		}()
495
	}
496
	close(start)
497
	wg.Wait()
498

499
	// Stop the web handler.
500
	cancel()
501

502
	select {
503
	case <-closed:
504
	case <-time.After(5 * time.Second):
505
		require.FailNow(t, "Server still running after 5 seconds.")
506
	}
507
}
508

509
// Test for availability of API endpoints in Prometheus Agent mode.
510
func TestAgentAPIEndPoints(t *testing.T) {
511
	t.Parallel()
512

513
	port := fmt.Sprintf(":%d", testutil.RandomUnprivilegedPort(t))
514

515
	opts := &Options{
516
		ListenAddress:  port,
517
		ReadTimeout:    30 * time.Second,
518
		MaxConnections: 512,
519
		Context:        nil,
520
		Storage:        nil,
521
		QueryEngine:    nil,
522
		ScrapeManager:  &scrape.Manager{},
523
		RuleManager:    &rules.Manager{},
524
		Notifier:       nil,
525
		RoutePrefix:    "/",
526
		EnableAdminAPI: true,
527
		ExternalURL: &url.URL{
528
			Scheme: "http",
529
			Host:   "localhost" + port,
530
			Path:   "/",
531
		},
532
		Version:  &PrometheusVersion{},
533
		Gatherer: prometheus.DefaultGatherer,
534
		IsAgent:  true,
535
	}
536

537
	opts.Flags = map[string]string{}
538

539
	webHandler := New(nil, opts)
540
	webHandler.SetReady(true)
541
	webHandler.config = &config.Config{}
542
	webHandler.notifier = &notifier.Manager{}
543
	l, err := webHandler.Listener()
544
	if err != nil {
545
		panic(fmt.Sprintf("Unable to start web listener: %s", err))
546
	}
547

548
	ctx, cancel := context.WithCancel(context.Background())
549
	defer cancel()
550
	go func() {
551
		err := webHandler.Run(ctx, l, "")
552
		if err != nil {
553
			panic(fmt.Sprintf("Can't start web handler:%s", err))
554
		}
555
	}()
556

557
	// Give some time for the web goroutine to run since we need the server
558
	// to be up before starting tests.
559
	time.Sleep(5 * time.Second)
560
	baseURL := "http://localhost" + port + "/api/v1"
561

562
	// Test for non-available endpoints in the Agent mode.
563
	for path, methods := range map[string][]string{
564
		"/labels":                      {http.MethodGet, http.MethodPost},
565
		"/label/:name/values":          {http.MethodGet},
566
		"/series":                      {http.MethodGet, http.MethodPost, http.MethodDelete},
567
		"/alertmanagers":               {http.MethodGet},
568
		"/query":                       {http.MethodGet, http.MethodPost},
569
		"/query_range":                 {http.MethodGet, http.MethodPost},
570
		"/query_exemplars":             {http.MethodGet, http.MethodPost},
571
		"/status/tsdb":                 {http.MethodGet},
572
		"/alerts":                      {http.MethodGet},
573
		"/rules":                       {http.MethodGet},
574
		"/admin/tsdb/delete_series":    {http.MethodPost, http.MethodPut},
575
		"/admin/tsdb/clean_tombstones": {http.MethodPost, http.MethodPut},
576
		"/admin/tsdb/snapshot":         {http.MethodPost, http.MethodPut},
577
	} {
578
		for _, m := range methods {
579
			req, err := http.NewRequest(m, baseURL+path, nil)
580
			require.NoError(t, err)
581
			resp, err := http.DefaultClient.Do(req)
582
			require.NoError(t, err)
583
			require.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode)
584
			t.Cleanup(func() {
585
				require.NoError(t, resp.Body.Close())
586
			})
587
		}
588
	}
589

590
	// Test for some of available endpoints in the Agent mode.
591
	for path, methods := range map[string][]string{
592
		"/targets":            {http.MethodGet},
593
		"/targets/metadata":   {http.MethodGet},
594
		"/metadata":           {http.MethodGet},
595
		"/status/config":      {http.MethodGet},
596
		"/status/runtimeinfo": {http.MethodGet},
597
		"/status/flags":       {http.MethodGet},
598
	} {
599
		for _, m := range methods {
600
			req, err := http.NewRequest(m, baseURL+path, nil)
601
			require.NoError(t, err)
602
			resp, err := http.DefaultClient.Do(req)
603
			require.NoError(t, err)
604
			require.Equal(t, http.StatusOK, resp.StatusCode)
605
			t.Cleanup(func() {
606
				require.NoError(t, resp.Body.Close())
607
			})
608
		}
609
	}
610
}
611

612
func cleanupTestResponse(t *testing.T, resp *http.Response) {
613
	_, err := io.Copy(io.Discard, resp.Body)
614
	require.NoError(t, err)
615
	require.NoError(t, resp.Body.Close())
616
}
617

618
func cleanupSnapshot(t *testing.T, dbDir string, resp *http.Response) {
619
	snapshot := &struct {
620
		Data struct {
621
			Name string `json:"name"`
622
		} `json:"data"`
623
	}{}
624
	b, err := io.ReadAll(resp.Body)
625
	require.NoError(t, err)
626
	require.NoError(t, json.Unmarshal(b, snapshot))
627
	require.NotZero(t, snapshot.Data.Name, "snapshot directory not returned")
628
	require.NoError(t, os.Remove(filepath.Join(dbDir, "snapshots", snapshot.Data.Name)))
629
	require.NoError(t, os.Remove(filepath.Join(dbDir, "snapshots")))
630
}
631