istio

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

15
package status
16

17
import (
18
	"context"
19
	"crypto/tls"
20
	"encoding/json"
21
	"errors"
22
	"fmt"
23
	"io"
24
	"net"
25
	"net/http"
26
	"net/http/httptest"
27
	"reflect"
28
	"strconv"
29
	"strings"
30
	"testing"
31
	"time"
32

33
	"github.com/prometheus/client_golang/prometheus"
34
	"github.com/prometheus/common/expfmt"
35
	"github.com/prometheus/prometheus/model/textparse"
36
	"go.uber.org/atomic"
37
	"google.golang.org/grpc"
38
	"google.golang.org/grpc/health"
39
	grpcHealth "google.golang.org/grpc/health/grpc_health_v1"
40
	"k8s.io/apimachinery/pkg/util/intstr"
41

42
	"istio.io/istio/pilot/cmd/pilot-agent/status/ready"
43
	"istio.io/istio/pilot/cmd/pilot-agent/status/testserver"
44
	"istio.io/istio/pkg/kube/apimirror"
45
	"istio.io/istio/pkg/lazy"
46
	"istio.io/istio/pkg/log"
47
	"istio.io/istio/pkg/test"
48
	"istio.io/istio/pkg/test/env"
49
	"istio.io/istio/pkg/test/util/assert"
50
	"istio.io/istio/pkg/test/util/retry"
51
)
52

53
type handler struct {
54
	// LastALPN stores the most recent ALPN requested. This is needed to determine info about a request,
55
	// since the appProber strips all headers/responses.
56
	lastAlpn *atomic.String
57
}
58

59
const (
60
	testHeader      = "Some-Header"
61
	testHeaderValue = "some-value"
62
	testHostValue   = "test.com:9999"
63
)
64

65
var liveServerStats = "cluster_manager.cds.update_success: 1\nlistener_manager.lds.update_success: 1\nserver.state: 0\nlistener_manager.workers_started: 1"
66

67
func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
68
	h.lastAlpn.Store(r.Proto)
69
	segments := strings.Split(r.URL.Path[1:], "/")
70
	switch segments[0] {
71
	case "header":
72
		if r.Host != testHostValue {
73
			log.Errorf("Missing expected host value %s, got %v", testHostValue, r.Host)
74
			w.WriteHeader(http.StatusBadRequest)
75
		}
76
		if r.Header.Get(testHeader) != testHeaderValue {
77
			log.Errorf("Missing expected Some-Header, got %v", r.Header)
78
			w.WriteHeader(http.StatusBadRequest)
79
		}
80
	case "redirect":
81
		http.Redirect(w, r, "/", http.StatusMovedPermanently)
82
	case "redirect-loop":
83
		http.Redirect(w, r, "/redirect-loop", http.StatusMovedPermanently)
84
	case "remote-redirect":
85
		http.Redirect(w, r, "http://example.com/foo", http.StatusMovedPermanently)
86
	case "", "hello/sunnyvale":
87
		w.Write([]byte("welcome, it works"))
88
	case "status":
89
		code, _ := strconv.Atoi(segments[1])
90
		w.Header().Set("Location", "/")
91
		w.WriteHeader(code)
92
	default:
93
		return
94
	}
95
}
96

97
func TestNewServer(t *testing.T) {
98
	testCases := []struct {
99
		probe string
100
		err   string
101
	}{
102
		// Json can't be parsed.
103
		{
104
			probe: "invalid-prober-json-encoding",
105
			err:   "failed to decode",
106
		},
107
		// map key is not well formed.
108
		{
109
			probe: `{"abc": {"path": "/app-foo/health"}}`,
110
			err:   "invalid path",
111
		},
112
		// invalid probe type
113
		{
114
			probe: `{"/app-health/hello-world/readyz": {"exec": {"command": [ "true" ]}}}`,
115
			err:   "invalid prober type",
116
		},
117
		// tcp probes are valid as well
118
		{
119
			probe: `{"/app-health/hello-world/readyz": {"tcpSocket": {"port": 8888}}}`,
120
		},
121
		// probes must be one of tcp, http or gRPC
122
		{
123
			probe: `{"/app-health/hello-world/readyz": {"tcpSocket": {"port": 8888}, "httpGet": {"path": "/", "port": 7777}}}`,
124
			err:   "must be one of type httpGet, tcpSocket or gRPC",
125
		},
126
		// probes must be one of tcp, http or gRPC
127
		{
128
			probe: `{"/app-health/hello-world/readyz": {"grpc": {"port": 8888}, "httpGet": {"path": "/", "port": 7777}}}`,
129
			err:   "must be one of type httpGet, tcpSocket or gRPC",
130
		},
131
		// Port is not Int typed (tcpSocket).
132
		{
133
			probe: `{"/app-health/hello-world/readyz": {"tcpSocket": {"port": "tcp"}}}`,
134
			err:   "must be int type",
135
		},
136
		// Port is not Int typed (httpGet).
137
		{
138
			probe: `{"/app-health/hello-world/readyz": {"httpGet": {"path": "/hello/sunnyvale", "port": "container-port-dontknow"}}}`,
139
			err:   "must be int type",
140
		},
141
		// A valid input.
142
		{
143
			probe: `{"/app-health/hello-world/readyz": {"httpGet": {"path": "/hello/sunnyvale", "port": 8080}},` +
144
				`"/app-health/business/livez": {"httpGet": {"path": "/buisiness/live", "port": 9090}}}`,
145
		},
146
		// long request timeout
147
		{
148
			probe: `{"/app-health/hello-world/readyz": {"httpGet": {"path": "/hello/sunnyvale", "port": 8080},` +
149
				`"initialDelaySeconds": 120,"timeoutSeconds": 10,"periodSeconds": 20}}`,
150
		},
151
		// A valid input with empty probing path, which happens when HTTPGetAction.Path is not specified.
152
		{
153
			probe: `{"/app-health/hello-world/readyz": {"httpGet": {"path": "/hello/sunnyvale", "port": 8080}},
154
"/app-health/business/livez": {"httpGet": {"port": 9090}}}`,
155
		},
156
		// A valid input without any prober info.
157
		{
158
			probe: `{}`,
159
		},
160
		// A valid input with probing path not starting with /, which happens when HTTPGetAction.Path does not start with a /.
161
		{
162
			probe: `{"/app-health/hello-world/readyz": {"httpGet": {"path": "hello/sunnyvale", "port": 8080}},
163
"/app-health/business/livez": {"httpGet": {"port": 9090}}}`,
164
		},
165
		// A valid gRPC probe.
166
		{
167
			probe: `{"/app-health/hello-world/readyz": {"gRPC": {"port": 8080}}}`,
168
		},
169
		// A valid gRPC probe with null service.
170
		{
171
			probe: `{"/app-health/hello-world/readyz": {"gRPC": {"port": 8080, "service": null}}}`,
172
		},
173
		// A valid gRPC probe with service.
174
		{
175
			probe: `{"/app-health/hello-world/readyz": {"gRPC": {"port": 8080, "service": "foo"}}}`,
176
		},
177
		// A valid gRPC probe with service and timeout.
178
		{
179
			probe: `{"/app-health/hello-world/readyz": {"gRPC": {"port": 8080, "service": "foo"}, "timeoutSeconds": 10}}`,
180
		},
181
	}
182
	for _, tc := range testCases {
183
		_, err := NewServer(Options{
184
			KubeAppProbers:     tc.probe,
185
			PrometheusRegistry: TestingRegistry(t),
186
		})
187

188
		if err == nil {
189
			if tc.err != "" {
190
				t.Errorf("test case failed [%v], expect error %v", tc.probe, tc.err)
191
			}
192
			continue
193
		}
194
		if tc.err == "" {
195
			t.Errorf("test case failed [%v], expect no error, got %v", tc.probe, err)
196
		}
197
		// error case, error string should match.
198
		if !strings.Contains(err.Error(), tc.err) {
199
			t.Errorf("test case failed [%v], expect error %v, got %v", tc.probe, tc.err, err)
200
		}
201
	}
202
}
203

204
func NewTestServer(t test.Failer, o Options) *Server {
205
	if o.PrometheusRegistry == nil {
206
		o.PrometheusRegistry = TestingRegistry(t)
207
	}
208
	server, err := NewServer(o)
209
	if err != nil {
210
		t.Fatalf("failed to create status server %v", err)
211
	}
212
	ctx, cancel := context.WithCancel(context.Background())
213
	t.Cleanup(cancel)
214
	go server.Run(ctx)
215

216
	if err := retry.UntilSuccess(func() error {
217
		server.mutex.RLock()
218
		statusPort := server.statusPort
219
		server.mutex.RUnlock()
220
		if statusPort == 0 {
221
			return fmt.Errorf("no port allocated")
222
		}
223
		return nil
224
	}, retry.Delay(time.Microsecond)); err != nil {
225
		t.Fatalf("failed to getport: %v", err)
226
	}
227

228
	return server
229
}
230

231
func TestPprof(t *testing.T) {
232
	pprofPath := "/debug/pprof/cmdline"
233
	// Starts the pilot agent status server.
234
	server := NewTestServer(t, Options{EnableProfiling: true})
235
	client := http.Client{}
236
	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost:%v/%s", server.statusPort, pprofPath), nil)
237
	if err != nil {
238
		t.Fatalf("[%v] failed to create request", pprofPath)
239
	}
240
	resp, err := client.Do(req)
241
	if err != nil {
242
		t.Fatal("request failed: ", err)
243
	}
244
	defer resp.Body.Close()
245
	if resp.StatusCode != http.StatusOK {
246
		t.Errorf("[%v] unexpected status code, want = %v, got = %v", pprofPath, http.StatusOK, resp.StatusCode)
247
	}
248
}
249

250
func TestStats(t *testing.T) {
251
	cases := []struct {
252
		name             string
253
		envoy            string
254
		app              string
255
		output           string
256
		expectParseError bool
257
	}{
258
		{
259
			name: "envoy metric only",
260
			envoy: `# TYPE my_metric counter
261
my_metric{} 0
262
# TYPE my_other_metric counter
263
my_other_metric{} 0
264
`,
265
			output: `# TYPE my_metric counter
266
my_metric{} 0
267
# TYPE my_other_metric counter
268
my_other_metric{} 0
269
`,
270
		},
271
		{
272
			name: "app metric only",
273
			app: `# TYPE my_metric counter
274
my_metric{} 0
275
# TYPE my_other_metric counter
276
my_other_metric{} 0
277
`,
278
			output: `# TYPE my_metric counter
279
my_metric{} 0
280
# TYPE my_other_metric counter
281
my_other_metric{} 0
282
`,
283
		},
284
		{
285
			name: "multiple metric",
286
			envoy: `# TYPE my_metric counter
287
my_metric{} 0
288
`,
289
			app: `# TYPE my_other_metric counter
290
my_other_metric{} 0
291
`,
292
			output: `# TYPE my_metric counter
293
my_metric{} 0
294
# TYPE my_other_metric counter
295
my_other_metric{} 0
296
`,
297
		},
298
		{
299
			name:  "agent metric",
300
			envoy: ``,
301
			app:   ``,
302
			// Agent metric is dynamic, so we just check a substring of it not the actual metric
303
			output: `
304
# TYPE istio_agent_scrapes_total counter
305
istio_agent_scrapes_total`,
306
		},
307
		// When the application and envoy share a metric, Prometheus will fail. This negative check validates this
308
		// assumption.
309
		{
310
			name: "conflict metric",
311
			envoy: `# TYPE my_metric counter
312
my_metric{} 0
313
# TYPE my_other_metric counter
314
my_other_metric{} 0
315
`,
316
			app: `# TYPE my_metric counter
317
my_metric{} 0
318
`,
319
			output: `# TYPE my_metric counter
320
my_metric{} 0
321
# TYPE my_other_metric counter
322
my_other_metric{} 0
323
# TYPE my_metric counter
324
my_metric{} 0
325
`,
326
			expectParseError: true,
327
		},
328
		{
329
			name: "conflict metric labeled",
330
			envoy: `# TYPE my_metric counter
331
my_metric{app="foo"} 0
332
`,
333
			app: `# TYPE my_metric counter
334
my_metric{app="bar"} 0
335
`,
336
			output: `# TYPE my_metric counter
337
my_metric{app="foo"} 0
338
# TYPE my_metric counter
339
my_metric{app="bar"} 0
340
`,
341
			expectParseError: true,
342
		},
343
	}
344
	for _, tt := range cases {
345
		t.Run(tt.name, func(t *testing.T) {
346
			rec := httptest.NewRecorder()
347
			envoy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
348
				if _, err := w.Write([]byte(tt.envoy)); err != nil {
349
					t.Fatalf("write failed: %v", err)
350
				}
351
			}))
352
			defer envoy.Close()
353
			app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
354
				if _, err := w.Write([]byte(tt.app)); err != nil {
355
					t.Fatalf("write failed: %v", err)
356
				}
357
			}))
358
			defer app.Close()
359
			envoyPort, err := strconv.Atoi(strings.Split(envoy.URL, ":")[2])
360
			if err != nil {
361
				t.Fatal(err)
362
			}
363
			server := &Server{
364
				prometheus: &PrometheusScrapeConfiguration{
365
					Port: strings.Split(app.URL, ":")[2],
366
				},
367
				envoyStatsPort: envoyPort,
368
				http:           &http.Client{},
369
				registry:       TestingRegistry(t),
370
			}
371
			req := &http.Request{}
372
			server.handleStats(rec, req)
373
			if rec.Code != 200 {
374
				t.Fatalf("handleStats() => %v; want 200", rec.Code)
375
			}
376
			if !strings.Contains(rec.Body.String(), tt.output) {
377
				t.Fatalf("handleStats() => %v; want %v", rec.Body.String(), tt.output)
378
			}
379

380
			parser := expfmt.TextParser{}
381
			mfMap, err := parser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
382
			if err != nil && !tt.expectParseError {
383
				t.Fatalf("failed to parse metrics: %v", err)
384
			} else if err == nil && tt.expectParseError {
385
				t.Fatalf("expected a prse error, got %+v", mfMap)
386
			}
387
		})
388
	}
389
}
390

391
func TestNegotiateMetricsFormat(t *testing.T) {
392
	cases := []struct {
393
		name        string
394
		contentType string
395
		expected    expfmt.Format
396
	}{
397
		{
398
			name:        "openmetrics minimal accept header",
399
			contentType: `application/openmetrics-text; version=0.0.1`,
400
			expected:    FmtOpenMetrics_0_0_1,
401
		},
402
		{
403
			name:        "openmetrics minimal v1 accept header",
404
			contentType: `application/openmetrics-text; version=1.0.0`,
405
			expected:    FmtOpenMetrics_1_0_0,
406
		},
407
		{
408
			name:        "openmetrics accept header",
409
			contentType: `application/openmetrics-text; version=0.0.1; charset=utf-8`,
410
			expected:    FmtOpenMetrics_0_0_1,
411
		},
412
		{
413
			name:        "openmetrics v1 accept header",
414
			contentType: `application/openmetrics-text; version=1.0.0; charset=utf-8`,
415
			expected:    FmtOpenMetrics_1_0_0,
416
		},
417
		{
418
			name:        "plaintext accept header",
419
			contentType: "text/plain; version=0.0.4; charset=utf-8",
420
			expected:    expfmt.NewFormat(expfmt.TypeTextPlain),
421
		},
422
		{
423
			name:        "empty accept header",
424
			contentType: "",
425
			expected:    expfmt.NewFormat(expfmt.TypeTextPlain),
426
		},
427
	}
428
	for _, tt := range cases {
429
		t.Run(tt.name, func(t *testing.T) {
430
			assert.Equal(t, negotiateMetricsFormat(tt.contentType), tt.expected)
431
		})
432
	}
433
}
434

435
func TestStatsContentType(t *testing.T) {
436
	appOpenMetrics := `# TYPE jvm info
437
# HELP jvm VM version info
438
jvm_info{runtime="OpenJDK Runtime Environment",vendor="AdoptOpenJDK",version="16.0.1+9"} 1.0
439
# TYPE jmx_config_reload_success counter
440
# HELP jmx_config_reload_success Number of times configuration have successfully been reloaded.
441
jmx_config_reload_success_total 0.0
442
jmx_config_reload_success_created 1.623984612719E9
443
# EOF
444
`
445
	appText004 := `# HELP jvm_info VM version info
446
# TYPE jvm_info gauge
447
jvm_info{runtime="OpenJDK Runtime Environment",vendor="AdoptOpenJDK",version="16.0.1+9",} 1.0
448
# HELP jmx_config_reload_failure_created Number of times configuration have failed to be reloaded.
449
# TYPE jmx_config_reload_failure_created gauge
450
jmx_config_reload_failure_created 1.624025983489E9
451
`
452
	envoy := `# TYPE my_metric counter
453
my_metric{} 0
454
# TYPE my_other_metric counter
455
my_other_metric{} 0
456
`
457
	cases := []struct {
458
		name             string
459
		acceptHeader     string
460
		expectParseError bool
461
	}{
462
		{
463
			name:         "openmetrics accept header",
464
			acceptHeader: `application/openmetrics-text; version=0.0.1,text/plain;version=0.0.4;q=0.5,*/*;q=0.1`,
465
		},
466
		{
467
			name:         "openmetrics v1 accept header",
468
			acceptHeader: `application/openmetrics-text; version=1.0.0,text/plain;version=0.0.4;q=0.5,*/*;q=0.1`,
469
		},
470
		{
471
			name:         "plaintext accept header",
472
			acceptHeader: string(FmtText),
473
		},
474
		{
475
			name:         "empty accept header",
476
			acceptHeader: "",
477
		},
478
	}
479
	for _, tt := range cases {
480
		t.Run(tt.name, func(t *testing.T) {
481
			rec := httptest.NewRecorder()
482
			envoy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
483
				if _, err := w.Write([]byte(envoy)); err != nil {
484
					t.Fatalf("write failed: %v", err)
485
				}
486
			}))
487
			defer envoy.Close()
488
			app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
489
				format := expfmt.NegotiateIncludingOpenMetrics(r.Header)
490
				var negotiatedMetrics string
491
				if strings.Contains(string(format), "text/plain") {
492
					negotiatedMetrics = appText004
493
				} else {
494
					negotiatedMetrics = appOpenMetrics
495
				}
496
				w.Header().Set("Content-Type", string(format))
497
				if _, err := w.Write([]byte(negotiatedMetrics)); err != nil {
498
					t.Fatalf("write failed: %v", err)
499
				}
500
			}))
501
			defer app.Close()
502
			envoyPort, err := strconv.Atoi(strings.Split(envoy.URL, ":")[2])
503
			if err != nil {
504
				t.Fatal(err)
505
			}
506
			server := &Server{
507
				prometheus: &PrometheusScrapeConfiguration{
508
					Port: strings.Split(app.URL, ":")[2],
509
				},
510
				registry:       TestingRegistry(t),
511
				envoyStatsPort: envoyPort,
512
				http:           &http.Client{},
513
			}
514
			req := &http.Request{}
515
			req.Header = make(http.Header)
516
			req.Header.Add("Accept", tt.acceptHeader)
517
			server.handleStats(rec, req)
518
			if rec.Code != 200 {
519
				t.Fatalf("handleStats() => %v; want 200", rec.Code)
520
			}
521

522
			if negotiateMetricsFormat(rec.Header().Get("Content-Type")) == FmtText {
523
				textParser := expfmt.TextParser{}
524
				_, err := textParser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
525
				if err != nil {
526
					t.Fatalf("failed to parse text metrics: %v", err)
527
				}
528
			} else {
529
				omParser := textparse.NewOpenMetricsParser(rec.Body.Bytes())
530
				for {
531
					_, err := omParser.Next()
532
					if err == io.EOF {
533
						break
534
					}
535
					if err != nil {
536
						t.Fatalf("failed to parse openmetrics: %v", err)
537
					}
538
				}
539
			}
540
		})
541
	}
542
}
543

544
func TestStatsError(t *testing.T) {
545
	fail := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
546
		w.WriteHeader(http.StatusInternalServerError)
547
	}))
548
	defer fail.Close()
549
	pass := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
550
		w.WriteHeader(http.StatusOK)
551
	}))
552
	defer pass.Close()
553
	failPort, err := strconv.Atoi(strings.Split(fail.URL, ":")[2])
554
	if err != nil {
555
		t.Fatal(err)
556
	}
557
	passPort, err := strconv.Atoi(strings.Split(pass.URL, ":")[2])
558
	if err != nil {
559
		t.Fatal(err)
560
	}
561
	cases := []struct {
562
		name  string
563
		envoy int
564
		app   int
565
	}{
566
		{"both pass", passPort, passPort},
567
		{"envoy pass", passPort, failPort},
568
		{"app pass", failPort, passPort},
569
		{"both fail", failPort, failPort},
570
	}
571
	for _, tt := range cases {
572
		t.Run(tt.name, func(t *testing.T) {
573
			server := &Server{
574
				prometheus: &PrometheusScrapeConfiguration{
575
					Port: strconv.Itoa(tt.app),
576
				},
577
				registry:       TestingRegistry(t),
578
				envoyStatsPort: tt.envoy,
579
				http:           &http.Client{},
580
			}
581
			req := &http.Request{}
582
			rec := httptest.NewRecorder()
583
			server.handleStats(rec, req)
584
			if rec.Code != 200 {
585
				t.Fatalf("handleStats() => %v; want 200", rec.Code)
586
			}
587
		})
588
	}
589
}
590

591
// initServerWithSize size is kB
592
func initServerWithSize(t *testing.B, size int) *Server {
593
	appText := `# TYPE jvm info
594
# HELP jvm VM version info
595
jvm_info{runtime="OpenJDK Runtime Environment",vendor="AdoptOpenJDK",version="16.0.1+9"} 1.0
596
# TYPE jmx_config_reload_success counter
597
# HELP jmx_config_reload_success Number of times configuration have successfully been reloaded.
598
jmx_config_reload_success_total 0.0
599
jmx_config_reload_success_created 1.623984612719E9
600
`
601
	appOpenMetrics := appText + "# EOF"
602

603
	envoy := strings.Builder{}
604
	envoy.Grow(size << 10 * 100)
605
	envoy.WriteString(`# TYPE my_metric counter
606
my_metric{} 0
607
# TYPE my_other_metric counter
608
my_other_metric{} 0
609
`)
610
	for i := 0; envoy.Len()+len(appText) < size<<10; i++ {
611
		envoy.WriteString("#TYPE my_other_metric_" + strconv.Itoa(i) + " counter\nmy_other_metric_" + strconv.Itoa(i) + " 0\n")
612
	}
613
	eb := []byte(envoy.String())
614

615
	envoyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
616
		if _, err := w.Write(eb); err != nil {
617
			t.Fatalf("write failed: %v", err)
618
		}
619
	}))
620
	t.Cleanup(envoyServer.Close)
621
	app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
622
		format := expfmt.NegotiateIncludingOpenMetrics(r.Header)
623
		var negotiatedMetrics string
624
		if format == FmtText {
625
			negotiatedMetrics = appText
626
		} else {
627
			negotiatedMetrics = appOpenMetrics
628
		}
629
		w.Header().Set("Content-Type", string(format))
630
		if _, err := w.Write([]byte(negotiatedMetrics)); err != nil {
631
			t.Fatalf("write failed: %v", err)
632
		}
633
	}))
634
	t.Cleanup(app.Close)
635
	envoyPort, err := strconv.Atoi(strings.Split(envoyServer.URL, ":")[2])
636
	if err != nil {
637
		t.Fatal(err)
638
	}
639
	registry, err := initializeMonitoring()
640
	if err != nil {
641
		t.Fatal(err)
642
	}
643
	server := &Server{
644
		registry: registry,
645
		prometheus: &PrometheusScrapeConfiguration{
646
			Port: strings.Split(app.URL, ":")[2],
647
		},
648
		envoyStatsPort: envoyPort,
649
		http:           &http.Client{},
650
	}
651
	t.ResetTimer()
652
	return server
653
}
654

655
func BenchmarkStats(t *testing.B) {
656
	tests := map[int]string{
657
		1:        "1kb",
658
		1 << 10:  "1mb",
659
		10 << 10: "10mb",
660
	}
661
	for size, v := range tests {
662
		server := initServerWithSize(t, size)
663
		t.Run("stats-fmttext-"+v, func(t *testing.B) {
664
			for i := 0; i < t.N; i++ {
665
				req := &http.Request{}
666
				req.Header = make(http.Header)
667
				req.Header.Add("Accept", string(FmtText))
668
				rec := httptest.NewRecorder()
669
				server.handleStats(rec, req)
670
			}
671
		})
672
		t.Run("stats-fmtopenmetrics-"+v, func(t *testing.B) {
673
			for i := 0; i < t.N; i++ {
674
				req := &http.Request{}
675
				req.Header = make(http.Header)
676
				req.Header.Add("Accept", string(FmtOpenMetrics_1_0_0))
677
				rec := httptest.NewRecorder()
678
				server.handleStats(rec, req)
679
			}
680
		})
681
	}
682
}
683

684
func TestAppProbe(t *testing.T) {
685
	// Starts the application first.
686
	listener, err := net.Listen("tcp", ":0")
687
	if err != nil {
688
		t.Errorf("failed to allocate unused port %v", err)
689
	}
690
	go http.Serve(listener, &handler{lastAlpn: atomic.NewString("")})
691
	appPort := listener.Addr().(*net.TCPAddr).Port
692

693
	simpleHTTPConfig := KubeAppProbers{
694
		"/app-health/hello-world/readyz": &Prober{
695
			HTTPGet: &apimirror.HTTPGetAction{
696
				Path: "/hello/sunnyvale",
697
				Port: intstr.IntOrString{IntVal: int32(appPort)},
698
			},
699
		},
700
		"/app-health/hello-world/livez": &Prober{
701
			HTTPGet: &apimirror.HTTPGetAction{
702
				Port: intstr.IntOrString{IntVal: int32(appPort)},
703
			},
704
		},
705
	}
706
	simpleTCPConfig := KubeAppProbers{
707
		"/app-health/hello-world/readyz": &Prober{
708
			TCPSocket: &apimirror.TCPSocketAction{
709
				Port: intstr.IntOrString{IntVal: int32(appPort)},
710
			},
711
		},
712
		"/app-health/hello-world/livez": &Prober{
713
			TCPSocket: &apimirror.TCPSocketAction{
714
				Port: intstr.IntOrString{IntVal: int32(appPort)},
715
			},
716
		},
717
	}
718

719
	type test struct {
720
		name       string
721
		probePath  string
722
		config     KubeAppProbers
723
		podIP      string
724
		ipv6       bool
725
		statusCode int
726
	}
727
	testCases := []test{
728
		{
729
			name:       "http-bad-path",
730
			probePath:  "bad-path-should-be-404",
731
			config:     simpleHTTPConfig,
732
			statusCode: http.StatusNotFound,
733
		},
734
		{
735
			name:       "http-readyz",
736
			probePath:  "app-health/hello-world/readyz",
737
			config:     simpleHTTPConfig,
738
			statusCode: http.StatusOK,
739
		},
740
		{
741
			name:       "http-livez",
742
			probePath:  "app-health/hello-world/livez",
743
			config:     simpleHTTPConfig,
744
			statusCode: http.StatusOK,
745
		},
746
		{
747
			name:       "http-livez-localhost",
748
			probePath:  "app-health/hello-world/livez",
749
			config:     simpleHTTPConfig,
750
			statusCode: http.StatusOK,
751
			podIP:      "localhost",
752
		},
753
		{
754
			name:      "http-readyz-header",
755
			probePath: "app-health/header/readyz",
756
			config: KubeAppProbers{
757
				"/app-health/header/readyz": &Prober{
758
					HTTPGet: &apimirror.HTTPGetAction{
759
						Port: intstr.IntOrString{IntVal: int32(appPort)},
760
						Path: "/header",
761
						HTTPHeaders: []apimirror.HTTPHeader{
762
							{Name: testHeader, Value: testHeaderValue},
763
							{Name: "Host", Value: testHostValue},
764
						},
765
					},
766
				},
767
			},
768
			statusCode: http.StatusOK,
769
		},
770
		{
771
			name:      "http-readyz-path",
772
			probePath: "app-health/hello-world/readyz",
773
			config: KubeAppProbers{
774
				"/app-health/hello-world/readyz": &Prober{
775
					HTTPGet: &apimirror.HTTPGetAction{
776
						Path: "hello/texas",
777
						Port: intstr.IntOrString{IntVal: int32(appPort)},
778
					},
779
				},
780
			},
781
			statusCode: http.StatusOK,
782
		},
783
		{
784
			name:      "http-livez-path",
785
			probePath: "app-health/hello-world/livez",
786
			config: KubeAppProbers{
787
				"/app-health/hello-world/livez": &Prober{
788
					HTTPGet: &apimirror.HTTPGetAction{
789
						Path: "hello/texas",
790
						Port: intstr.IntOrString{IntVal: int32(appPort)},
791
					},
792
				},
793
			},
794
			statusCode: http.StatusOK,
795
		},
796
		{
797
			name:       "tcp-readyz",
798
			probePath:  "app-health/hello-world/readyz",
799
			config:     simpleTCPConfig,
800
			statusCode: http.StatusOK,
801
		},
802
		{
803
			name:       "tcp-livez",
804
			probePath:  "app-health/hello-world/livez",
805
			config:     simpleTCPConfig,
806
			statusCode: http.StatusOK,
807
		},
808
		{
809
			name:       "tcp-livez-ipv4",
810
			probePath:  "app-health/hello-world/livez",
811
			config:     simpleTCPConfig,
812
			statusCode: http.StatusOK,
813
			podIP:      "127.0.0.1",
814
		},
815
		{
816
			name:       "tcp-livez-ipv6",
817
			probePath:  "app-health/hello-world/livez",
818
			config:     simpleTCPConfig,
819
			statusCode: http.StatusOK,
820
			podIP:      "::1",
821
			ipv6:       true,
822
		},
823
		{
824
			name:       "tcp-livez-wrapped-ipv6",
825
			probePath:  "app-health/hello-world/livez",
826
			config:     simpleTCPConfig,
827
			statusCode: http.StatusInternalServerError,
828
			podIP:      "[::1]",
829
			ipv6:       true,
830
		},
831
		{
832
			name:       "tcp-livez-localhost",
833
			probePath:  "app-health/hello-world/livez",
834
			config:     simpleTCPConfig,
835
			statusCode: http.StatusOK,
836
			podIP:      "localhost",
837
		},
838
		{
839
			name:      "redirect",
840
			probePath: "app-health/redirect/livez",
841
			config: KubeAppProbers{
842
				"/app-health/redirect/livez": &Prober{
843
					HTTPGet: &apimirror.HTTPGetAction{
844
						Path: "redirect",
845
						Port: intstr.IntOrString{IntVal: int32(appPort)},
846
					},
847
				},
848
			},
849
			statusCode: http.StatusOK,
850
		},
851
		{
852
			name:      "redirect loop",
853
			probePath: "app-health/redirect-loop/livez",
854
			config: KubeAppProbers{
855
				"/app-health/redirect-loop/livez": &Prober{
856
					HTTPGet: &apimirror.HTTPGetAction{
857
						Path: "redirect-loop",
858
						Port: intstr.IntOrString{IntVal: int32(appPort)},
859
					},
860
				},
861
			},
862
			statusCode: http.StatusInternalServerError,
863
		},
864
		{
865
			name:      "remote redirect",
866
			probePath: "app-health/remote-redirect/livez",
867
			config: KubeAppProbers{
868
				"/app-health/remote-redirect/livez": &Prober{
869
					HTTPGet: &apimirror.HTTPGetAction{
870
						Path: "remote-redirect",
871
						Port: intstr.IntOrString{IntVal: int32(appPort)},
872
					},
873
				},
874
			},
875
			statusCode: http.StatusOK,
876
		},
877
	}
878
	testFn := func(t *testing.T, tc test) {
879
		appProber, err := json.Marshal(tc.config)
880
		if err != nil {
881
			t.Fatalf("invalid app probers")
882
		}
883
		config := Options{
884
			KubeAppProbers: string(appProber),
885
			PodIP:          tc.podIP,
886
			IPv6:           tc.ipv6,
887
		}
888
		server := NewTestServer(t, config)
889
		// Starts the pilot agent status server.
890
		if tc.ipv6 {
891
			server.upstreamLocalAddress = &net.TCPAddr{IP: net.ParseIP("::1")} // required because ::6 is NOT a loopback address (IPv6 only has ::1)
892
		}
893

894
		client := http.Client{}
895
		req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost:%v/%s", server.statusPort, tc.probePath), nil)
896
		if err != nil {
897
			t.Fatalf("[%v] failed to create request", tc.probePath)
898
		}
899
		if c := tc.config["/"+tc.probePath]; c != nil {
900
			if hc := c.HTTPGet; hc != nil {
901
				for _, h := range hc.HTTPHeaders {
902
					req.Header[h.Name] = append(req.Header[h.Name], h.Value)
903
				}
904
			}
905
		}
906
		// This is simulating the kubelet behavior of setting the Host to Header["Host"].
907
		// https://github.com/kubernetes/kubernetes/blob/d3b7391dc2f1040083ee2a8bfcb02edf7b0ded4b/pkg/probe/http/request.go#L84C1-L84C1
908
		req.Host = req.Header.Get("Host")
909
		resp, err := client.Do(req)
910
		if err != nil {
911
			t.Fatal("request failed: ", err)
912
		}
913
		defer resp.Body.Close()
914
		if resp.StatusCode != tc.statusCode {
915
			t.Errorf("[%v] unexpected status code, want = %v, got = %v", tc.probePath, tc.statusCode, resp.StatusCode)
916
		}
917
	}
918
	for _, tc := range testCases {
919
		t.Run(tc.name, func(t *testing.T) { testFn(t, tc) })
920
	}
921
	// Next we check ever
922
	t.Run("status codes", func(t *testing.T) {
923
		for code := http.StatusOK; code <= http.StatusNetworkAuthenticationRequired; code++ {
924
			if http.StatusText(code) == "" { // Not a valid HTTP code
925
				continue
926
			}
927
			expect := code
928
			if isRedirect(code) {
929
				expect = 200
930
			}
931
			t.Run(fmt.Sprint(code), func(t *testing.T) {
932
				testFn(t, test{
933
					probePath: "app-health/code/livez",
934
					config: KubeAppProbers{
935
						"/app-health/code/livez": &Prober{
936
							TimeoutSeconds: 1,
937
							HTTPGet: &apimirror.HTTPGetAction{
938
								Path: fmt.Sprintf("status/%d", code),
939
								Port: intstr.IntOrString{IntVal: int32(appPort)},
940
							},
941
						},
942
					},
943
					statusCode: expect,
944
				})
945
			})
946
		}
947
	})
948
}
949

950
func TestHttpsAppProbe(t *testing.T) {
951
	setupServer := func(t *testing.T, alpn []string) (uint16, func() string) {
952
		// Starts the application first.
953
		listener, err := net.Listen("tcp", ":0")
954
		if err != nil {
955
			t.Errorf("failed to allocate unused port %v", err)
956
		}
957
		t.Cleanup(func() { listener.Close() })
958
		keyFile := env.IstioSrc + "/pilot/cmd/pilot-agent/status/test-cert/cert.key"
959
		crtFile := env.IstioSrc + "/pilot/cmd/pilot-agent/status/test-cert/cert.crt"
960
		cert, err := tls.LoadX509KeyPair(crtFile, keyFile)
961
		if err != nil {
962
			t.Fatalf("could not load TLS keys: %v", err)
963
		}
964
		serverTLSConfig := &tls.Config{
965
			Certificates: []tls.Certificate{cert},
966
			NextProtos:   alpn,
967
			MinVersion:   tls.VersionTLS12,
968
		}
969
		tlsListener := tls.NewListener(listener, serverTLSConfig)
970
		h := &handler{lastAlpn: atomic.NewString("")}
971
		srv := http.Server{Handler: h}
972
		go srv.Serve(tlsListener)
973
		appPort := listener.Addr().(*net.TCPAddr).Port
974

975
		// Starts the pilot agent status server.
976
		server := NewTestServer(t, Options{
977
			KubeAppProbers: fmt.Sprintf(`{"/app-health/hello-world/readyz": {"httpGet": {"path": "/hello/sunnyvale", "port": %v, "scheme": "HTTPS"}},
978
"/app-health/hello-world/livez": {"httpGet": {"port": %v, "scheme": "HTTPS"}}}`, appPort, appPort),
979
		})
980
		return server.statusPort, h.lastAlpn.Load
981
	}
982
	testCases := []struct {
983
		name             string
984
		probePath        string
985
		expectedProtocol string
986
		statusCode       int
987
		alpns            []string
988
	}{
989
		{
990
			name:       "bad-path-should-be-disallowed",
991
			probePath:  "bad-path-should-be-disallowed",
992
			statusCode: http.StatusNotFound,
993
		},
994
		{
995
			name:             "readyz",
996
			probePath:        "app-health/hello-world/readyz",
997
			statusCode:       http.StatusOK,
998
			expectedProtocol: "HTTP/1.1",
999
			alpns:            nil,
1000
		},
1001
		{
1002
			name:             "livez",
1003
			probePath:        "app-health/hello-world/livez",
1004
			statusCode:       http.StatusOK,
1005
			expectedProtocol: "HTTP/1.1",
1006
		},
1007
		{
1008
			name:             "h1 only",
1009
			probePath:        "app-health/hello-world/readyz",
1010
			statusCode:       http.StatusOK,
1011
			expectedProtocol: "HTTP/1.1",
1012
			alpns:            []string{"http/1.1"},
1013
		},
1014
		{
1015
			name:             "h2 only",
1016
			probePath:        "app-health/hello-world/readyz",
1017
			statusCode:       http.StatusOK,
1018
			expectedProtocol: "HTTP/2.0",
1019
			alpns:            []string{"h2"},
1020
		},
1021
		{
1022
			name:             "prefer h2",
1023
			probePath:        "app-health/hello-world/readyz",
1024
			statusCode:       http.StatusOK,
1025
			expectedProtocol: "HTTP/2.0",
1026
			alpns:            []string{"h2", "http/1.1"},
1027
		},
1028
		{
1029
			name:             "prefer h1",
1030
			probePath:        "app-health/hello-world/readyz",
1031
			statusCode:       http.StatusOK,
1032
			expectedProtocol: "HTTP/2.0",
1033
			alpns:            []string{"h2", "http/1.1"},
1034
		},
1035
		{
1036
			name:       "unknown alpn",
1037
			probePath:  "app-health/hello-world/readyz",
1038
			statusCode: http.StatusInternalServerError,
1039
			alpns:      []string{"foo"},
1040
		},
1041
	}
1042
	for _, tc := range testCases {
1043
		t.Run(tc.name, func(t *testing.T) {
1044
			statusPort, getAlpn := setupServer(t, tc.alpns)
1045
			client := http.Client{}
1046
			req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost:%d/%s", statusPort, tc.probePath), nil)
1047
			if err != nil {
1048
				t.Fatalf("failed to create request")
1049
			}
1050
			resp, err := client.Do(req)
1051
			if err != nil {
1052
				t.Fatal("request failed")
1053
			}
1054
			defer resp.Body.Close()
1055
			if resp.StatusCode != tc.statusCode {
1056
				t.Errorf("unexpected status code, want = %v, got = %v", tc.statusCode, resp.StatusCode)
1057
			}
1058
			if got := getAlpn(); got != tc.expectedProtocol {
1059
				t.Errorf("unexpected protocol, want = %v, got = %v", tc.expectedProtocol, got)
1060
			}
1061
		})
1062
	}
1063
}
1064

1065
func TestGRPCAppProbe(t *testing.T) {
1066
	appServer := grpc.NewServer()
1067
	healthServer := health.NewServer()
1068
	healthServer.SetServingStatus("serving-svc", grpcHealth.HealthCheckResponse_SERVING)
1069
	healthServer.SetServingStatus("unknown-svc", grpcHealth.HealthCheckResponse_UNKNOWN)
1070
	healthServer.SetServingStatus("not-serving-svc", grpcHealth.HealthCheckResponse_NOT_SERVING)
1071
	grpcHealth.RegisterHealthServer(appServer, healthServer)
1072

1073
	listener, err := net.Listen("tcp", ":0")
1074
	if err != nil {
1075
		t.Errorf("failed to allocate unused port %v", err)
1076
	}
1077
	go appServer.Serve(listener)
1078
	defer appServer.GracefulStop()
1079

1080
	appPort := listener.Addr().(*net.TCPAddr).Port
1081
	// Starts the pilot agent status server.
1082
	server := NewTestServer(t, Options{
1083
		KubeAppProbers: fmt.Sprintf(`
1084
{
1085
    "/app-health/foo/livez": {
1086
        "grpc": {
1087
            "port": %v, 
1088
            "service": null
1089
        }, 
1090
        "timeoutSeconds": 1
1091
    }, 
1092
    "/app-health/foo/readyz": {
1093
        "grpc": {
1094
            "port": %v, 
1095
            "service": "not-serving-svc"
1096
        }, 
1097
        "timeoutSeconds": 1
1098
    }, 
1099
    "/app-health/bar/livez": {
1100
        "grpc": {
1101
            "port": %v, 
1102
            "service": "serving-svc"
1103
        }, 
1104
        "timeoutSeconds": 10
1105
    }, 
1106
    "/app-health/bar/readyz": {
1107
        "grpc": {
1108
            "port": %v, 
1109
            "service": "unknown-svc"
1110
        }, 
1111
        "timeoutSeconds": 10
1112
    }
1113
}`, appPort, appPort, appPort, appPort),
1114
	})
1115
	statusPort := server.statusPort
1116
	t.Logf("status server starts at port %v, app starts at port %v", statusPort, appPort)
1117

1118
	testCases := []struct {
1119
		name       string
1120
		probePath  string
1121
		statusCode int
1122
	}{
1123
		{
1124
			name:       "bad-path-should-be-disallowed",
1125
			probePath:  fmt.Sprintf(":%v/bad-path-should-be-disallowed", statusPort),
1126
			statusCode: http.StatusNotFound,
1127
		},
1128
		{
1129
			name:       "foo-livez",
1130
			probePath:  fmt.Sprintf(":%v/app-health/foo/livez", statusPort),
1131
			statusCode: http.StatusOK,
1132
		},
1133
		{
1134
			name:       "foo-readyz",
1135
			probePath:  fmt.Sprintf(":%v/app-health/foo/readyz", statusPort),
1136
			statusCode: http.StatusInternalServerError,
1137
		},
1138
		{
1139
			name:       "bar-livez",
1140
			probePath:  fmt.Sprintf(":%v/app-health/bar/livez", statusPort),
1141
			statusCode: http.StatusOK,
1142
		},
1143
		{
1144
			name:       "bar-readyz",
1145
			probePath:  fmt.Sprintf(":%v/app-health/bar/readyz", statusPort),
1146
			statusCode: http.StatusInternalServerError,
1147
		},
1148
	}
1149
	for _, tc := range testCases {
1150
		t.Run(tc.name, func(t *testing.T) {
1151
			client := http.Client{}
1152
			req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost%s", tc.probePath), nil)
1153
			if err != nil {
1154
				t.Errorf("[%v] failed to create request", tc.probePath)
1155
			}
1156
			resp, err := client.Do(req)
1157
			if err != nil {
1158
				t.Fatal("request failed")
1159
			}
1160
			defer resp.Body.Close()
1161
			if resp.StatusCode != tc.statusCode {
1162
				t.Errorf("[%v] unexpected status code, want = %v, got = %v", tc.probePath, tc.statusCode, resp.StatusCode)
1163
			}
1164
		})
1165
	}
1166
}
1167

1168
func TestGRPCAppProbeWithIPV6(t *testing.T) {
1169
	appServer := grpc.NewServer()
1170
	healthServer := health.NewServer()
1171
	healthServer.SetServingStatus("serving-svc", grpcHealth.HealthCheckResponse_SERVING)
1172
	healthServer.SetServingStatus("unknown-svc", grpcHealth.HealthCheckResponse_UNKNOWN)
1173
	healthServer.SetServingStatus("not-serving-svc", grpcHealth.HealthCheckResponse_NOT_SERVING)
1174
	grpcHealth.RegisterHealthServer(appServer, healthServer)
1175

1176
	listener, err := net.Listen("tcp", ":0")
1177
	if err != nil {
1178
		t.Errorf("failed to allocate unused port %v", err)
1179
	}
1180
	go appServer.Serve(listener)
1181
	defer appServer.GracefulStop()
1182

1183
	appPort := listener.Addr().(*net.TCPAddr).Port
1184
	// Starts the pilot agent status server.
1185
	server := NewTestServer(t, Options{
1186
		IPv6:  true,
1187
		PodIP: "::1",
1188
		KubeAppProbers: fmt.Sprintf(`
1189
{
1190
    "/app-health/foo/livez": {
1191
        "grpc": {
1192
            "port": %v, 
1193
            "service": null
1194
        }, 
1195
        "timeoutSeconds": 1
1196
    }, 
1197
    "/app-health/foo/readyz": {
1198
        "grpc": {
1199
            "port": %v, 
1200
            "service": "not-serving-svc"
1201
        }, 
1202
        "timeoutSeconds": 1
1203
    }, 
1204
    "/app-health/bar/livez": {
1205
        "grpc": {
1206
            "port": %v, 
1207
            "service": "serving-svc"
1208
        }, 
1209
        "timeoutSeconds": 10
1210
    }, 
1211
    "/app-health/bar/readyz": {
1212
        "grpc": {
1213
            "port": %v, 
1214
            "service": "unknown-svc"
1215
        }, 
1216
        "timeoutSeconds": 10
1217
    }
1218
}`, appPort, appPort, appPort, appPort),
1219
	})
1220

1221
	server.upstreamLocalAddress = &net.TCPAddr{IP: net.ParseIP("::1")} // required because ::6 is NOT a loopback address (IPv6 only has ::1)
1222

1223
	testCases := []struct {
1224
		name       string
1225
		probePath  string
1226
		statusCode int
1227
	}{
1228
		{
1229
			name:       "bad-path-should-be-disallowed",
1230
			probePath:  fmt.Sprintf(":%v/bad-path-should-be-disallowed", server.statusPort),
1231
			statusCode: http.StatusNotFound,
1232
		},
1233
		{
1234
			name:       "foo-livez",
1235
			probePath:  fmt.Sprintf(":%v/app-health/foo/livez", server.statusPort),
1236
			statusCode: http.StatusOK,
1237
		},
1238
		{
1239
			name:       "foo-readyz",
1240
			probePath:  fmt.Sprintf(":%v/app-health/foo/readyz", server.statusPort),
1241
			statusCode: http.StatusInternalServerError,
1242
		},
1243
		{
1244
			name:       "bar-livez",
1245
			probePath:  fmt.Sprintf(":%v/app-health/bar/livez", server.statusPort),
1246
			statusCode: http.StatusOK,
1247
		},
1248
		{
1249
			name:       "bar-readyz",
1250
			probePath:  fmt.Sprintf(":%v/app-health/bar/readyz", server.statusPort),
1251
			statusCode: http.StatusInternalServerError,
1252
		},
1253
	}
1254
	for _, tc := range testCases {
1255
		t.Run(tc.name, func(t *testing.T) {
1256
			client := http.Client{}
1257
			req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost%s", tc.probePath), nil)
1258
			if err != nil {
1259
				t.Errorf("[%v] failed to create request", tc.probePath)
1260
			}
1261
			resp, err := client.Do(req)
1262
			if err != nil {
1263
				t.Fatal("request failed")
1264
			}
1265
			defer resp.Body.Close()
1266
			if resp.StatusCode != tc.statusCode {
1267
				t.Errorf("[%v] unexpected status code, want = %v, got = %v", tc.probePath, tc.statusCode, resp.StatusCode)
1268
			}
1269
		})
1270
	}
1271
}
1272

1273
func TestProbeHeader(t *testing.T) {
1274
	headerChecker := func(t *testing.T, header http.Header) net.Listener {
1275
		listener, err := net.Listen("tcp", ":0")
1276
		if err != nil {
1277
			t.Fatalf("failed to allocate unused port %v", err)
1278
		}
1279
		go http.Serve(listener, http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
1280
			r.Header.Del("User-Agent")
1281
			r.Header.Del("Accept-Encoding")
1282
			if !reflect.DeepEqual(r.Header, header) {
1283
				t.Errorf("unexpected header, want = %v, got = %v", header, r.Header)
1284
				http.Error(rw, "", http.StatusBadRequest)
1285
				return
1286
			}
1287
			http.Error(rw, "", http.StatusOK)
1288
		}))
1289
		return listener
1290
	}
1291

1292
	testCases := []struct {
1293
		name          string
1294
		originHeaders http.Header
1295
		proxyHeaders  []apimirror.HTTPHeader
1296
		want          http.Header
1297
	}{
1298
		{
1299
			name: "Only Origin",
1300
			originHeaders: http.Header{
1301
				testHeader: []string{testHeaderValue},
1302
			},
1303
			proxyHeaders: []apimirror.HTTPHeader{},
1304
			want: http.Header{
1305
				testHeader:   []string{testHeaderValue},
1306
				"Connection": []string{"close"},
1307
			},
1308
		},
1309
		{
1310
			name: "Only Origin, has multiple values",
1311
			originHeaders: http.Header{
1312
				testHeader: []string{testHeaderValue, testHeaderValue},
1313
			},
1314
			proxyHeaders: []apimirror.HTTPHeader{},
1315
			want: http.Header{
1316
				testHeader:   []string{testHeaderValue, testHeaderValue},
1317
				"Connection": []string{"close"},
1318
			},
1319
		},
1320
		{
1321
			name:          "Only Proxy",
1322
			originHeaders: http.Header{},
1323
			proxyHeaders: []apimirror.HTTPHeader{
1324
				{
1325
					Name:  testHeader,
1326
					Value: testHeaderValue,
1327
				},
1328
			},
1329
			want: http.Header{
1330
				"Connection": []string{"close"},
1331
			},
1332
		},
1333
		{
1334
			name:          "Only Proxy, has multiple values",
1335
			originHeaders: http.Header{},
1336
			proxyHeaders: []apimirror.HTTPHeader{
1337
				{
1338
					Name:  testHeader,
1339
					Value: testHeaderValue,
1340
				},
1341
				{
1342
					Name:  testHeader,
1343
					Value: testHeaderValue,
1344
				},
1345
			},
1346
			want: http.Header{
1347
				"Connection": []string{"close"},
1348
			},
1349
		},
1350
		{
1351
			name: "Proxy overwrites Origin",
1352
			originHeaders: http.Header{
1353
				testHeader: []string{testHeaderValue},
1354
			},
1355
			proxyHeaders: []apimirror.HTTPHeader{
1356
				{
1357
					Name:  testHeader,
1358
					Value: testHeaderValue + "over",
1359
				},
1360
			},
1361
			want: http.Header{
1362
				testHeader:   []string{testHeaderValue},
1363
				"Connection": []string{"close"},
1364
			},
1365
		},
1366
	}
1367
	for _, tc := range testCases {
1368
		t.Run(tc.name, func(t *testing.T) {
1369
			svc := headerChecker(t, tc.want)
1370
			defer svc.Close()
1371
			probePath := "/app-health/hello-world/livez"
1372
			appAddress := svc.Addr().(*net.TCPAddr)
1373
			appProber, err := json.Marshal(KubeAppProbers{
1374
				probePath: &Prober{
1375
					HTTPGet: &apimirror.HTTPGetAction{
1376
						Port:        intstr.IntOrString{IntVal: int32(appAddress.Port)},
1377
						Host:        appAddress.IP.String(),
1378
						Path:        "/header",
1379
						HTTPHeaders: tc.proxyHeaders,
1380
					},
1381
				},
1382
			})
1383
			if err != nil {
1384
				t.Fatalf("invalid app probers")
1385
			}
1386
			config := Options{
1387
				KubeAppProbers: string(appProber),
1388
			}
1389
			// Starts the pilot agent status server.
1390
			server := NewTestServer(t, config)
1391
			client := http.Client{}
1392
			req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost:%v%s", server.statusPort, probePath), nil)
1393
			if err != nil {
1394
				t.Fatal("failed to create request: ", err)
1395
			}
1396
			req.Header = tc.originHeaders
1397
			resp, err := client.Do(req)
1398
			if err != nil {
1399
				t.Fatal("request failed: ", err)
1400
			}
1401
			defer resp.Body.Close()
1402
			if resp.StatusCode != http.StatusOK {
1403
				t.Errorf("unexpected status code, want = %v, got = %v", http.StatusOK, resp.StatusCode)
1404
			}
1405
		})
1406
	}
1407
}
1408

1409
func TestHandleQuit(t *testing.T) {
1410
	tests := []struct {
1411
		name       string
1412
		method     string
1413
		remoteAddr string
1414
		expected   int
1415
	}{
1416
		{
1417
			name:       "should send a sigterm for valid requests",
1418
			method:     "POST",
1419
			remoteAddr: "127.0.0.1",
1420
			expected:   http.StatusOK,
1421
		},
1422
		{
1423
			name:       "should send a sigterm for valid ipv6 requests",
1424
			method:     "POST",
1425
			remoteAddr: "[::1]",
1426
			expected:   http.StatusOK,
1427
		},
1428
		{
1429
			name:       "should require POST method",
1430
			method:     "GET",
1431
			remoteAddr: "127.0.0.1",
1432
			expected:   http.StatusMethodNotAllowed,
1433
		},
1434
		{
1435
			name:     "should require localhost",
1436
			method:   "POST",
1437
			expected: http.StatusForbidden,
1438
		},
1439
	}
1440

1441
	for _, tt := range tests {
1442
		t.Run(tt.name, func(t *testing.T) {
1443
			shutdown := make(chan struct{})
1444
			s := NewTestServer(t, Options{
1445
				Shutdown: func() {
1446
					close(shutdown)
1447
				},
1448
			})
1449
			req, err := http.NewRequest(tt.method, "/quitquitquit", nil)
1450
			if err != nil {
1451
				t.Fatal(err)
1452
			}
1453

1454
			if tt.remoteAddr != "" {
1455
				req.RemoteAddr = tt.remoteAddr + ":" + fmt.Sprint(s.statusPort)
1456
			}
1457

1458
			resp := httptest.NewRecorder()
1459
			s.handleQuit(resp, req)
1460
			if resp.Code != tt.expected {
1461
				t.Fatalf("Expected response code %v got %v", tt.expected, resp.Code)
1462
			}
1463

1464
			if tt.expected == http.StatusOK {
1465
				select {
1466
				case <-shutdown:
1467
				case <-time.After(time.Second):
1468
					t.Fatalf("Failed to receive expected shutdown")
1469
				}
1470
			} else {
1471
				select {
1472
				case <-shutdown:
1473
					t.Fatalf("unexpected shutdown")
1474
				default:
1475
				}
1476
			}
1477
		})
1478
	}
1479
}
1480

1481
func TestAdditionalProbes(t *testing.T) {
1482
	rp := readyProbe{}
1483
	urp := unreadyProbe{}
1484
	testCases := []struct {
1485
		name   string
1486
		probes []ready.Prober
1487
		err    error
1488
	}{
1489
		{
1490
			name:   "success probe",
1491
			probes: []ready.Prober{rp},
1492
			err:    nil,
1493
		},
1494
		{
1495
			name:   "not ready probe",
1496
			probes: []ready.Prober{urp},
1497
			err:    errors.New("not ready"),
1498
		},
1499
		{
1500
			name:   "both probes",
1501
			probes: []ready.Prober{rp, urp},
1502
			err:    errors.New("not ready"),
1503
		},
1504
	}
1505
	testServer := testserver.CreateAndStartServer(liveServerStats)
1506
	defer testServer.Close()
1507
	for _, tc := range testCases {
1508
		server, err := NewServer(Options{
1509
			Probes:    tc.probes,
1510
			AdminPort: uint16(testServer.Listener.Addr().(*net.TCPAddr).Port),
1511
		})
1512
		if err != nil {
1513
			t.Errorf("failed to construct server")
1514
		}
1515
		err = server.isReady()
1516
		if tc.err == nil {
1517
			if err != nil {
1518
				t.Errorf("Unexpected result, expected: %v got: %v", tc.err, err)
1519
			}
1520
		} else {
1521
			if err.Error() != tc.err.Error() {
1522
				t.Errorf("Unexpected result, expected: %v got: %v", tc.err, err)
1523
			}
1524
		}
1525

1526
	}
1527
}
1528

1529
type readyProbe struct{}
1530

1531
func (s readyProbe) Check() error {
1532
	return nil
1533
}
1534

1535
type unreadyProbe struct{}
1536

1537
func (u unreadyProbe) Check() error {
1538
	return errors.New("not ready")
1539
}
1540

1541
var reg = lazy.New(initializeMonitoring)
1542

1543
func TestingRegistry(t test.Failer) prometheus.Gatherer {
1544
	r, err := reg.Get()
1545
	if err != nil {
1546
		t.Fatal(err)
1547
	}
1548
	return r
1549
}
1550