2016-11-02 21:41:50 +03:00
|
|
|
// Copyright 2016 The etcd Authors
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
|
|
|
package integration
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bufio"
|
2017-09-07 00:57:25 +03:00
|
|
|
"context"
|
2016-11-02 21:41:50 +03:00
|
|
|
"io"
|
|
|
|
"net"
|
|
|
|
"net/http"
|
|
|
|
"strconv"
|
|
|
|
"strings"
|
|
|
|
"testing"
|
|
|
|
"time"
|
|
|
|
|
2018-08-29 03:13:10 +03:00
|
|
|
"go.etcd.io/etcd/clientv3"
|
|
|
|
"go.etcd.io/etcd/integration"
|
|
|
|
"go.etcd.io/etcd/pkg/testutil"
|
|
|
|
"go.etcd.io/etcd/pkg/transport"
|
2016-11-02 21:41:50 +03:00
|
|
|
|
2017-02-03 22:05:59 +03:00
|
|
|
grpcprom "github.com/grpc-ecosystem/go-grpc-prometheus"
|
2017-10-26 05:27:22 +03:00
|
|
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
2017-02-03 22:05:59 +03:00
|
|
|
"google.golang.org/grpc"
|
2016-11-02 21:41:50 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
func TestV3ClientMetrics(t *testing.T) {
|
|
|
|
defer testutil.AfterTest(t)
|
|
|
|
|
|
|
|
var (
|
2018-07-22 01:48:53 +03:00
|
|
|
addr = "localhost:27989"
|
2016-11-02 21:41:50 +03:00
|
|
|
ln net.Listener
|
|
|
|
err error
|
|
|
|
)
|
|
|
|
|
2017-11-10 21:08:48 +03:00
|
|
|
// listen for all Prometheus metrics
|
2016-11-02 21:41:50 +03:00
|
|
|
donec := make(chan struct{})
|
|
|
|
go func() {
|
|
|
|
defer close(donec)
|
|
|
|
|
2017-10-26 05:27:22 +03:00
|
|
|
srv := &http.Server{Handler: promhttp.Handler()}
|
2016-11-02 21:41:50 +03:00
|
|
|
srv.SetKeepAlivesEnabled(false)
|
|
|
|
|
|
|
|
ln, err = transport.NewUnixListener(addr)
|
|
|
|
if err != nil {
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Fatalf("Error: %v occurred while listening on addr: %v", err, addr)
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
err = srv.Serve(ln)
|
2017-04-13 20:48:53 +03:00
|
|
|
if err != nil && !transport.IsClosedConnError(err) {
|
2016-11-02 21:41:50 +03:00
|
|
|
t.Fatalf("Err serving http requests: %v", err)
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
url := "unix://" + addr + "/metrics"
|
|
|
|
|
2017-10-25 00:57:33 +03:00
|
|
|
clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1, SkipCreatingClient: true})
|
2016-11-02 21:41:50 +03:00
|
|
|
defer clus.Terminate(t)
|
|
|
|
|
2017-02-03 22:05:59 +03:00
|
|
|
cfg := clientv3.Config{
|
|
|
|
Endpoints: []string{clus.Members[0].GRPCAddr()},
|
|
|
|
DialOptions: []grpc.DialOption{
|
|
|
|
grpc.WithUnaryInterceptor(grpcprom.UnaryClientInterceptor),
|
|
|
|
grpc.WithStreamInterceptor(grpcprom.StreamClientInterceptor),
|
|
|
|
},
|
|
|
|
}
|
|
|
|
cli, cerr := clientv3.New(cfg)
|
|
|
|
if cerr != nil {
|
|
|
|
t.Fatal(cerr)
|
|
|
|
}
|
|
|
|
defer cli.Close()
|
|
|
|
|
|
|
|
wc := cli.Watch(context.Background(), "foo")
|
2016-11-02 21:41:50 +03:00
|
|
|
|
|
|
|
wBefore := sumCountersForMetricAndLabels(t, url, "grpc_client_msg_received_total", "Watch", "bidi_stream")
|
|
|
|
|
|
|
|
pBefore := sumCountersForMetricAndLabels(t, url, "grpc_client_started_total", "Put", "unary")
|
|
|
|
|
2017-02-03 22:05:59 +03:00
|
|
|
_, err = cli.Put(context.Background(), "foo", "bar")
|
2016-11-02 21:41:50 +03:00
|
|
|
if err != nil {
|
|
|
|
t.Errorf("Error putting value in key store")
|
|
|
|
}
|
|
|
|
|
|
|
|
pAfter := sumCountersForMetricAndLabels(t, url, "grpc_client_started_total", "Put", "unary")
|
|
|
|
if pBefore+1 != pAfter {
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Errorf("grpc_client_started_total expected %d, got %d", 1, pAfter-pBefore)
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// consume watch response
|
|
|
|
select {
|
|
|
|
case <-wc:
|
|
|
|
case <-time.After(10 * time.Second):
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Error("Timeout occurred for getting watch response")
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
wAfter := sumCountersForMetricAndLabels(t, url, "grpc_client_msg_received_total", "Watch", "bidi_stream")
|
|
|
|
if wBefore+1 != wAfter {
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Errorf("grpc_client_msg_received_total expected %d, got %d", 1, wAfter-wBefore)
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
ln.Close()
|
|
|
|
<-donec
|
|
|
|
}
|
|
|
|
|
|
|
|
func sumCountersForMetricAndLabels(t *testing.T, url string, metricName string, matchingLabelValues ...string) int {
|
|
|
|
count := 0
|
|
|
|
for _, line := range getHTTPBodyAsLines(t, url) {
|
|
|
|
ok := true
|
|
|
|
if !strings.HasPrefix(line, metricName) {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, labelValue := range matchingLabelValues {
|
|
|
|
if !strings.Contains(line, `"`+labelValue+`"`) {
|
|
|
|
ok = false
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if !ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
valueString := line[strings.LastIndex(line, " ")+1 : len(line)-1]
|
|
|
|
valueFloat, err := strconv.ParseFloat(valueString, 32)
|
|
|
|
if err != nil {
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Fatalf("failed parsing value for line: %v and matchingLabelValues: %v", line, matchingLabelValues)
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
count += int(valueFloat)
|
|
|
|
}
|
|
|
|
return count
|
|
|
|
}
|
|
|
|
|
|
|
|
func getHTTPBodyAsLines(t *testing.T, url string) []string {
|
|
|
|
cfgtls := transport.TLSInfo{}
|
|
|
|
tr, err := transport.NewTransport(cfgtls, time.Second)
|
|
|
|
if err != nil {
|
2016-11-08 23:37:33 +03:00
|
|
|
t.Fatalf("Error getting transport: %v", err)
|
2016-11-02 21:41:50 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
tr.MaxIdleConns = -1
|
|
|
|
tr.DisableKeepAlives = true
|
|
|
|
|
|
|
|
cli := &http.Client{Transport: tr}
|
|
|
|
|
|
|
|
resp, err := cli.Get(url)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("Error fetching: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
reader := bufio.NewReader(resp.Body)
|
|
|
|
lines := []string{}
|
|
|
|
for {
|
|
|
|
line, err := reader.ReadString('\n')
|
|
|
|
if err != nil {
|
|
|
|
if err == io.EOF {
|
|
|
|
break
|
|
|
|
} else {
|
|
|
|
t.Fatalf("error reading: %v", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
lines = append(lines, line)
|
|
|
|
}
|
|
|
|
resp.Body.Close()
|
|
|
|
return lines
|
|
|
|
}
|