prometheus

1
// Copyright 2015 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 consul
15

16
import (
17
	"context"
18
	"net/http"
19
	"net/http/httptest"
20
	"net/url"
21
	"testing"
22
	"time"
23

24
	"github.com/go-kit/log"
25
	"github.com/prometheus/client_golang/prometheus"
26
	"github.com/prometheus/common/config"
27
	"github.com/prometheus/common/model"
28
	"github.com/stretchr/testify/require"
29
	"go.uber.org/goleak"
30
	"gopkg.in/yaml.v2"
31

32
	"github.com/prometheus/prometheus/discovery"
33
	"github.com/prometheus/prometheus/discovery/targetgroup"
34
)
35

36
func TestMain(m *testing.M) {
37
	goleak.VerifyTestMain(m)
38
}
39

40
// TODO: Add ability to unregister metrics?
41
func NewTestMetrics(t *testing.T, conf discovery.Config, reg prometheus.Registerer) discovery.DiscovererMetrics {
42
	refreshMetrics := discovery.NewRefreshMetrics(reg)
43
	require.NoError(t, refreshMetrics.Register())
44

45
	metrics := conf.NewDiscovererMetrics(prometheus.NewRegistry(), refreshMetrics)
46
	require.NoError(t, metrics.Register())
47

48
	return metrics
49
}
50

51
func TestConfiguredService(t *testing.T) {
52
	conf := &SDConfig{
53
		Services: []string{"configuredServiceName"},
54
	}
55

56
	metrics := NewTestMetrics(t, conf, prometheus.NewRegistry())
57

58
	consulDiscovery, err := NewDiscovery(conf, nil, metrics)
59
	require.NoError(t, err, "when initializing discovery")
60
	require.True(t, consulDiscovery.shouldWatch("configuredServiceName", []string{""}),
61
		"Expected service %s to be watched", "configuredServiceName")
62
	require.False(t, consulDiscovery.shouldWatch("nonConfiguredServiceName", []string{""}),
63
		"Expected service %s to not be watched", "nonConfiguredServiceName")
64
}
65

66
func TestConfiguredServiceWithTag(t *testing.T) {
67
	conf := &SDConfig{
68
		Services:    []string{"configuredServiceName"},
69
		ServiceTags: []string{"http"},
70
	}
71

72
	metrics := NewTestMetrics(t, conf, prometheus.NewRegistry())
73

74
	consulDiscovery, err := NewDiscovery(conf, nil, metrics)
75
	require.NoError(t, err, "when initializing discovery")
76
	require.False(t, consulDiscovery.shouldWatch("configuredServiceName", []string{""}),
77
		"Expected service %s to not be watched without tag", "configuredServiceName")
78

79
	require.True(t, consulDiscovery.shouldWatch("configuredServiceName", []string{"http"}),
80
		"Expected service %s to be watched with tag %s", "configuredServiceName", "http")
81

82
	require.False(t, consulDiscovery.shouldWatch("nonConfiguredServiceName", []string{""}),
83
		"Expected service %s to not be watched without tag", "nonConfiguredServiceName")
84

85
	require.False(t, consulDiscovery.shouldWatch("nonConfiguredServiceName", []string{"http"}),
86
		"Expected service %s to not be watched with tag %s", "nonConfiguredServiceName", "http")
87
}
88

89
func TestConfiguredServiceWithTags(t *testing.T) {
90
	type testcase struct {
91
		// What we've configured to watch.
92
		conf *SDConfig
93
		// The service we're checking if we should watch or not.
94
		serviceName string
95
		serviceTags []string
96
		shouldWatch bool
97
	}
98

99
	cases := []testcase{
100
		{
101
			conf: &SDConfig{
102
				Services:    []string{"configuredServiceName"},
103
				ServiceTags: []string{"http", "v1"},
104
			},
105
			serviceName: "configuredServiceName",
106
			serviceTags: []string{""},
107
			shouldWatch: false,
108
		},
109
		{
110
			conf: &SDConfig{
111
				Services:    []string{"configuredServiceName"},
112
				ServiceTags: []string{"http", "v1"},
113
			},
114
			serviceName: "configuredServiceName",
115
			serviceTags: []string{"http", "v1"},
116
			shouldWatch: true,
117
		},
118
		{
119
			conf: &SDConfig{
120
				Services:    []string{"configuredServiceName"},
121
				ServiceTags: []string{"http", "v1"},
122
			},
123
			serviceName: "nonConfiguredServiceName",
124
			serviceTags: []string{""},
125
			shouldWatch: false,
126
		},
127
		{
128
			conf: &SDConfig{
129
				Services:    []string{"configuredServiceName"},
130
				ServiceTags: []string{"http", "v1"},
131
			},
132
			serviceName: "nonConfiguredServiceName",
133
			serviceTags: []string{"http, v1"},
134
			shouldWatch: false,
135
		},
136
		{
137
			conf: &SDConfig{
138
				Services:    []string{"configuredServiceName"},
139
				ServiceTags: []string{"http", "v1"},
140
			},
141
			serviceName: "configuredServiceName",
142
			serviceTags: []string{"http", "v1", "foo"},
143
			shouldWatch: true,
144
		},
145
		{
146
			conf: &SDConfig{
147
				Services:    []string{"configuredServiceName"},
148
				ServiceTags: []string{"http", "v1", "foo"},
149
			},
150
			serviceName: "configuredServiceName",
151
			serviceTags: []string{"http", "v1", "foo"},
152
			shouldWatch: true,
153
		},
154
		{
155
			conf: &SDConfig{
156
				Services:    []string{"configuredServiceName"},
157
				ServiceTags: []string{"http", "v1"},
158
			},
159
			serviceName: "configuredServiceName",
160
			serviceTags: []string{"http", "v1", "v1"},
161
			shouldWatch: true,
162
		},
163
	}
164

165
	for _, tc := range cases {
166
		metrics := NewTestMetrics(t, tc.conf, prometheus.NewRegistry())
167

168
		consulDiscovery, err := NewDiscovery(tc.conf, nil, metrics)
169
		require.NoError(t, err, "when initializing discovery")
170
		ret := consulDiscovery.shouldWatch(tc.serviceName, tc.serviceTags)
171
		require.Equal(t, tc.shouldWatch, ret, "Watched service and tags: %s %+v, input was %s %+v",
172
			tc.conf.Services, tc.conf.ServiceTags, tc.serviceName, tc.serviceTags)
173
	}
174
}
175

176
func TestNonConfiguredService(t *testing.T) {
177
	conf := &SDConfig{}
178

179
	metrics := NewTestMetrics(t, conf, prometheus.NewRegistry())
180

181
	consulDiscovery, err := NewDiscovery(conf, nil, metrics)
182
	require.NoError(t, err, "when initializing discovery")
183
	require.True(t, consulDiscovery.shouldWatch("nonConfiguredServiceName", []string{""}), "Expected service %s to be watched", "nonConfiguredServiceName")
184
}
185

186
const (
187
	AgentAnswer       = `{"Config": {"Datacenter": "test-dc"}}`
188
	ServiceTestAnswer = `
189
[{
190
	"Node": {
191
		"ID": "b78c2e48-5ef3-1814-31b8-0d880f50471e",
192
		"Node": "node1",
193
		"Address": "1.1.1.1",
194
		"Datacenter": "test-dc",
195
		"TaggedAddresses": {
196
			"lan": "192.168.10.10",
197
			"wan": "10.0.10.10"
198
		},
199
		"Meta": {"rack_name": "2304"},
200
		"CreateIndex": 1,
201
		"ModifyIndex": 1
202
	},
203
	"Service": {
204
		"ID": "test",
205
		"Service": "test",
206
		"Tags": ["tag1"],
207
		"Address": "",
208
		"Meta": {"version":"1.0.0","environment":"staging"},
209
		"Port": 3341,
210
		"Weights": {
211
			"Passing": 1,
212
			"Warning": 1
213
		},
214
		"EnableTagOverride": false,
215
		"ProxyDestination": "",
216
		"Proxy": {},
217
		"Connect": {},
218
		"CreateIndex": 1,
219
		"ModifyIndex": 1
220
	},
221
	"Checks": [{
222
		"Node": "node1",
223
		"CheckID": "serfHealth",
224
		"Name": "Serf Health Status",
225
		"Status": "passing"
226
	}]
227
}]`
228

229
	ServicesTestAnswer = `{"test": ["tag1"], "other": ["tag2"]}`
230
)
231

232
func newServer(t *testing.T) (*httptest.Server, *SDConfig) {
233
	// github.com/hashicorp/consul/testutil/ would be nice but it needs a local consul binary.
234
	stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
235
		response := ""
236
		switch r.URL.String() {
237
		case "/v1/agent/self":
238
			response = AgentAnswer
239
		case "/v1/health/service/test?node-meta=rack_name%3A2304&stale=&tag=tag1&wait=120000ms":
240
			response = ServiceTestAnswer
241
		case "/v1/health/service/test?wait=120000ms":
242
			response = ServiceTestAnswer
243
		case "/v1/health/service/other?wait=120000ms":
244
			response = `[]`
245
		case "/v1/catalog/services?node-meta=rack_name%3A2304&stale=&wait=120000ms":
246
			response = ServicesTestAnswer
247
		case "/v1/catalog/services?wait=120000ms":
248
			response = ServicesTestAnswer
249
		case "/v1/catalog/services?index=1&node-meta=rack_name%3A2304&stale=&wait=120000ms":
250
			time.Sleep(5 * time.Second)
251
			response = ServicesTestAnswer
252
		case "/v1/catalog/services?index=1&wait=120000ms":
253
			time.Sleep(5 * time.Second)
254
			response = ServicesTestAnswer
255
		default:
256
			t.Errorf("Unhandled consul call: %s", r.URL)
257
		}
258
		w.Header().Add("X-Consul-Index", "1")
259
		w.Write([]byte(response))
260
	}))
261
	stuburl, err := url.Parse(stub.URL)
262
	require.NoError(t, err)
263

264
	config := &SDConfig{
265
		Server:          stuburl.Host,
266
		Token:           "fake-token",
267
		RefreshInterval: model.Duration(1 * time.Second),
268
	}
269
	return stub, config
270
}
271

272
func newDiscovery(t *testing.T, config *SDConfig) *Discovery {
273
	logger := log.NewNopLogger()
274

275
	metrics := NewTestMetrics(t, config, prometheus.NewRegistry())
276

277
	d, err := NewDiscovery(config, logger, metrics)
278
	require.NoError(t, err)
279
	return d
280
}
281

282
func checkOneTarget(t *testing.T, tg []*targetgroup.Group) {
283
	require.Len(t, tg, 1)
284
	target := tg[0]
285
	require.Equal(t, "test-dc", string(target.Labels["__meta_consul_dc"]))
286
	require.Equal(t, target.Source, string(target.Labels["__meta_consul_service"]))
287
	if target.Source == "test" {
288
		// test service should have one node.
289
		require.NotEmpty(t, target.Targets, "Test service should have one node")
290
	}
291
}
292

293
// Watch all the services in the catalog.
294
func TestAllServices(t *testing.T) {
295
	stub, config := newServer(t)
296
	defer stub.Close()
297

298
	d := newDiscovery(t, config)
299

300
	ctx, cancel := context.WithCancel(context.Background())
301
	ch := make(chan []*targetgroup.Group)
302
	go func() {
303
		d.Run(ctx, ch)
304
		close(ch)
305
	}()
306
	checkOneTarget(t, <-ch)
307
	checkOneTarget(t, <-ch)
308
	cancel()
309
	<-ch
310
}
311

312
// targetgroup with no targets is emitted if no services were discovered.
313
func TestNoTargets(t *testing.T) {
314
	stub, config := newServer(t)
315
	defer stub.Close()
316
	config.ServiceTags = []string{"missing"}
317

318
	d := newDiscovery(t, config)
319

320
	ctx, cancel := context.WithCancel(context.Background())
321
	ch := make(chan []*targetgroup.Group)
322
	go func() {
323
		d.Run(ctx, ch)
324
		close(ch)
325
	}()
326

327
	targets := (<-ch)[0].Targets
328
	require.Empty(t, targets)
329
	cancel()
330
	<-ch
331
}
332

333
// Watch only the test service.
334
func TestOneService(t *testing.T) {
335
	stub, config := newServer(t)
336
	defer stub.Close()
337

338
	config.Services = []string{"test"}
339
	d := newDiscovery(t, config)
340

341
	ctx, cancel := context.WithCancel(context.Background())
342
	ch := make(chan []*targetgroup.Group)
343
	go d.Run(ctx, ch)
344
	checkOneTarget(t, <-ch)
345
	cancel()
346
}
347

348
// Watch the test service with a specific tag and node-meta.
349
func TestAllOptions(t *testing.T) {
350
	stub, config := newServer(t)
351
	defer stub.Close()
352

353
	config.Services = []string{"test"}
354
	config.NodeMeta = map[string]string{"rack_name": "2304"}
355
	config.ServiceTags = []string{"tag1"}
356
	config.AllowStale = true
357
	config.Token = "fake-token"
358

359
	d := newDiscovery(t, config)
360

361
	ctx, cancel := context.WithCancel(context.Background())
362
	ch := make(chan []*targetgroup.Group)
363
	go func() {
364
		d.Run(ctx, ch)
365
		close(ch)
366
	}()
367
	checkOneTarget(t, <-ch)
368
	cancel()
369
	<-ch
370
}
371

372
func TestGetDatacenterShouldReturnError(t *testing.T) {
373
	for _, tc := range []struct {
374
		handler    func(http.ResponseWriter, *http.Request)
375
		errMessage string
376
	}{
377
		{
378
			// Define a handler that will return status 500.
379
			handler: func(w http.ResponseWriter, r *http.Request) {
380
				w.WriteHeader(http.StatusInternalServerError)
381
			},
382
			errMessage: "Unexpected response code: 500 ()",
383
		},
384
		{
385
			// Define a handler that will return incorrect response.
386
			handler: func(w http.ResponseWriter, r *http.Request) {
387
				w.Write([]byte(`{"Config": {"Not-Datacenter": "test-dc"}}`))
388
			},
389
			errMessage: "invalid value '<nil>' for Config.Datacenter",
390
		},
391
	} {
392
		stub := httptest.NewServer(http.HandlerFunc(tc.handler))
393
		stuburl, err := url.Parse(stub.URL)
394
		require.NoError(t, err)
395

396
		config := &SDConfig{
397
			Server:          stuburl.Host,
398
			Token:           "fake-token",
399
			RefreshInterval: model.Duration(1 * time.Second),
400
		}
401
		defer stub.Close()
402
		d := newDiscovery(t, config)
403

404
		// Should be empty if not initialized.
405
		require.Equal(t, "", d.clientDatacenter)
406

407
		err = d.getDatacenter()
408

409
		// An error should be returned.
410
		require.Equal(t, tc.errMessage, err.Error())
411
		// Should still be empty.
412
		require.Equal(t, "", d.clientDatacenter)
413
	}
414
}
415

416
func TestUnmarshalConfig(t *testing.T) {
417
	unmarshal := func(d []byte) func(interface{}) error {
418
		return func(o interface{}) error {
419
			return yaml.Unmarshal(d, o)
420
		}
421
	}
422

423
	goodConfig := DefaultSDConfig
424
	goodConfig.Username = "123"
425
	goodConfig.Password = "1234"
426
	goodConfig.HTTPClientConfig = config.HTTPClientConfig{
427
		BasicAuth: &config.BasicAuth{
428
			Username: "123",
429
			Password: "1234",
430
		},
431
		FollowRedirects: true,
432
		EnableHTTP2:     true,
433
	}
434

435
	cases := []struct {
436
		name       string
437
		config     string
438
		expected   SDConfig
439
		errMessage string
440
	}{
441
		{
442
			name: "good",
443
			config: `
444
server: localhost:8500
445
username: 123
446
password: 1234
447
`,
448
			expected: goodConfig,
449
		},
450
		{
451
			name: "username and password and basic auth configured",
452
			config: `
453
server: localhost:8500
454
username: 123
455
password: 1234
456
basic_auth:
457
  username: 12345
458
  password: 123456
459
`,
460
			errMessage: "at most one of consul SD configuration username and password and basic auth can be configured",
461
		},
462
		{
463
			name: "token and authorization configured",
464
			config: `
465
server: localhost:8500
466
token: 1234567
467
authorization:
468
  credentials: 12345678
469
`,
470
			errMessage: "at most one of consul SD token, authorization, or oauth2 can be configured",
471
		},
472
		{
473
			name: "token and oauth2 configured",
474
			config: `
475
server: localhost:8500
476
token: 1234567
477
oauth2:
478
  client_id: 10
479
  client_secret: 11
480
  token_url: http://example.com
481
`,
482
			errMessage: "at most one of consul SD token, authorization, or oauth2 can be configured",
483
		},
484
	}
485

486
	for _, test := range cases {
487
		t.Run(test.name, func(t *testing.T) {
488
			var config SDConfig
489
			err := config.UnmarshalYAML(unmarshal([]byte(test.config)))
490
			if err != nil {
491
				require.EqualError(t, err, test.errMessage)
492
				return
493
			}
494
			require.Empty(t, test.errMessage, "Expected error.")
495

496
			require.Equal(t, test.expected, config)
497
		})
498
	}
499
}
500