blob: 2a963e0581e98b35c968626f5ecc75b75ed65fa1 [file] [log] [blame]
// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package main
import (
"fmt"
"io"
"math"
"strings"
cloudmonitoring "google.golang.org/api/cloudmonitoring/v2beta2"
"v.io/jiri/tool"
"v.io/v23"
"v.io/v23/context"
"v.io/v23/naming"
"v.io/v23/options"
"v.io/v23/rpc"
"v.io/v23/services/stats"
"v.io/v23/vdl"
"v.io/x/devtools/internal/monitoring"
"v.io/x/devtools/internal/test"
)
// Human-readable service names.
const (
snMounttable = "mounttable"
snApplications = "application repository"
snBinaries = "binary repository"
snIdentity = "identity service"
snMacaroon = "macaroon service"
snGoogleIdentity = "google identity service"
snBinaryDischarger = "binary discharger"
snRole = "role service"
snProxy = "proxy service"
snGroups = "groups service"
hostnameStatSuffix = "__debug/stats/system/hostname"
zoneStatSuffix = "__debug/stats/system/gce/zone"
)
// serviceMountedNames is a map from human-readable service names to their
// relative mounted names in the global mounttable.
var serviceMountedNames = map[string]string{
snMounttable: "",
snApplications: "applications",
snBinaries: "binaries",
snIdentity: "identity/dev.v.io:u",
snMacaroon: "identity/dev.v.io:u/macaroon",
snGoogleIdentity: "identity/dev.v.io:u/google",
snBinaryDischarger: "identity/dev.v.io:u/discharger",
snRole: "identity/role",
snProxy: "proxy-mon",
snGroups: "groups",
}
type aggregator struct {
data []float64
min float64
max float64
sum float64
}
func newAggregator() *aggregator {
return &aggregator{
data: []float64{},
min: math.MaxFloat64,
}
}
func (a *aggregator) add(v float64) {
a.data = append(a.data, v)
a.min = math.Min(a.min, v)
a.max = math.Max(a.max, v)
a.sum += v
}
func (a *aggregator) avg() float64 {
return a.sum / float64(len(a.data))
}
func (a *aggregator) count() float64 {
return float64(len(a.data))
}
func (a *aggregator) String() string {
return fmt.Sprintf("min: %f, max: %f, avg: %f", a.min, a.max, a.avg())
}
type statValue struct {
name string
value interface{}
}
func (sv *statValue) getStringValue() string {
return fmt.Sprint(sv.value)
}
func (sv *statValue) getFloat64Value() (float64, error) {
switch i := sv.value.(type) {
case float64:
return i, nil
case int64:
return float64(i), nil
default:
return 0, fmt.Errorf("invalid value: %v", sv.value)
}
}
func getMountedName(serviceName string) (string, error) {
relativeName, ok := serviceMountedNames[serviceName]
if !ok {
return "", fmt.Errorf("service name %q not found", serviceName)
}
return fmt.Sprintf("%s/%s", namespaceRootFlag, relativeName), nil
}
// getStat gets the given stat using rpc.
func getStat(v23ctx *context.T, ctx *tool.Context, me naming.MountEntry, pattern string) ([]*statValue, error) {
v23ctx, cancel := context.WithTimeout(v23ctx, timeout)
defer cancel()
call, err := v23.GetClient(v23ctx).StartCall(v23ctx, "", rpc.GlobMethod, []interface{}{pattern}, options.Preresolved{&me})
if err != nil {
return nil, err
}
hasErrors := false
ret := []*statValue{}
mountEntryName := me.Name
for {
var gr naming.GlobReply
err := call.Recv(&gr)
if err == io.EOF {
break
}
if err != nil {
return nil, err
}
switch v := gr.(type) {
case naming.GlobReplyEntry:
me.Name = naming.Join(mountEntryName, v.Value.Name)
value, err := stats.StatsClient("").Value(v23ctx, options.Preresolved{&me})
if err != nil {
fmt.Fprintf(ctx.Stderr(), "Failed to get stat: %v\n%v\n", v.Value.Name, err)
hasErrors = true
continue
}
var convertedValue interface{}
if err := vdl.Convert(&convertedValue, value); err != nil {
fmt.Fprintf(ctx.Stderr(), "Failed to convert value for %v: %v\n", v.Value.Name, err)
hasErrors = true
continue
}
ret = append(ret, &statValue{
name: v.Value.Name,
value: convertedValue,
})
case naming.GlobReplyError:
fmt.Fprintf(ctx.Stderr(), "Glob failed at %q: %v", v.Value.Name, v.Value.Error)
}
}
if hasErrors || len(ret) == 0 {
return nil, fmt.Errorf("failed to get stat")
}
if err := call.Finish(); err != nil {
return nil, err
}
return ret, nil
}
// resolveAndProcessServiceName resolves the given service name and groups the
// result entries by their routing ids.
func resolveAndProcessServiceName(v23ctx *context.T, ctx *tool.Context, serviceName, serviceMountedName string) (map[string]naming.MountEntry, error) {
// Resolve the name.
v23ctx, cancel := context.WithTimeout(v23ctx, timeout)
defer cancel()
ns := v23.GetNamespace(v23ctx)
entry, err := ns.ShallowResolve(v23ctx, serviceMountedName)
if err != nil {
return nil, err
}
resolvedNames := []string{}
for _, server := range entry.Servers {
fullName := naming.JoinAddressName(server.Server, entry.Name)
resolvedNames = append(resolvedNames, fullName)
}
// Group resolved names by their routing ids.
groups := map[string]naming.MountEntry{}
if serviceName == snMounttable {
// Mounttable resolves to itself, so we just use a dummy routing id with
// its original mounted name.
groups["-"] = naming.MountEntry{
Servers: []naming.MountedServer{naming.MountedServer{Server: serviceMountedName}},
}
} else {
for _, resolvedName := range resolvedNames {
serverName, relativeName := naming.SplitAddressName(resolvedName)
ep, err := v23.NewEndpoint(serverName)
if err != nil {
return nil, err
}
routingId := ep.RoutingID().String()
if _, ok := groups[routingId]; !ok {
groups[routingId] = naming.MountEntry{}
}
curMountEntry := groups[routingId]
curMountEntry.Servers = append(curMountEntry.Servers, naming.MountedServer{Server: serverName})
// resolvedNames are resolved from the same service so they should have
// the same relative name.
curMountEntry.Name = relativeName
groups[routingId] = curMountEntry
}
}
return groups, nil
}
// getServiceLocation returns the given service's location (instance and zone).
// If the service is replicated, the instance name is the pod name.
//
// To make it simpler and faster, we look up service's location in hard-coded "zone maps"
// for both non-replicated and replicated services.
func getServiceLocation(v23ctx *context.T, ctx *tool.Context, me naming.MountEntry) (*monitoring.ServiceLocation, error) {
// Check "__debug/stats/system/metadata/hostname" stat to get service's
// host name.
me.Name = ""
hostnameResult, err := getStat(v23ctx, ctx, me, hostnameStatSuffix)
if err != nil {
return nil, err
}
hostname := hostnameResult[0].getStringValue()
// Check "__debug/stats/system/gce/zone" stat to get service's
// zone name.
zoneResult, err := getStat(v23ctx, ctx, me, zoneStatSuffix)
if err != nil {
return nil, err
}
zone := zoneResult[0].getStringValue()
// The zone stat exported by services is in the form of:
// projects/632758215260/zones/us-central1-c
// We only need the last part.
parts := strings.Split(zone, "/")
zone = parts[len(parts)-1]
return &monitoring.ServiceLocation{
Instance: hostname,
Zone: zone,
}, nil
}
// sendDataToGCM sends the given metric to Google Cloud Monitoring.
func sendDataToGCM(s *cloudmonitoring.Service, md *cloudmonitoring.MetricDescriptor, value float64, now, instance, zone string, extraLabelKeys ...string) error {
// Sending value 0 will cause error.
if math.Abs(value) < 1e-7 {
return nil
}
labels := []string{}
if instance != "" {
labels = append(labels, instance)
}
if zone != "" {
labels = append(labels, zone)
}
for _, key := range extraLabelKeys {
labels = append(labels, key)
}
if len(labels) != len(md.Labels) {
return fmt.Errorf("wrong number of label keys: want %d, got %d", len(md.Labels), len(labels))
}
labelsMap := map[string]string{}
for i := range labels {
labelsMap[md.Labels[i].Key] = labels[i]
}
if _, err := s.Timeseries.Write(projectFlag, &cloudmonitoring.WriteTimeseriesRequest{
Timeseries: []*cloudmonitoring.TimeseriesPoint{
&cloudmonitoring.TimeseriesPoint{
Point: &cloudmonitoring.Point{
DoubleValue: value,
Start: now,
End: now,
},
TimeseriesDesc: &cloudmonitoring.TimeseriesDescriptor{
Metric: md.Name,
Labels: labelsMap,
},
},
},
}).Do(); err != nil {
return fmt.Errorf("Timeseries Write failed for metric %q with value %q: %v", md.Name, value, err)
}
return nil
}
func sendAggregatedDataToGCM(ctx *tool.Context, s *cloudmonitoring.Service, md *cloudmonitoring.MetricDescriptor, agg *aggregator, now string, extraLabelKeys ...string) error {
labels := []string{}
for _, l := range extraLabelKeys {
labels = append(labels, l)
}
minLabels := append(labels, "min")
if err := sendDataToGCM(s, md, agg.min, now, "", "", minLabels...); err != nil {
return err
}
maxLabels := append(labels, "max")
if err := sendDataToGCM(s, md, agg.max, now, "", "", maxLabels...); err != nil {
return err
}
avgLabels := append(labels, "avg")
if err := sendDataToGCM(s, md, agg.avg(), now, "", "", avgLabels...); err != nil {
return err
}
sumLabels := append(labels, "sum")
if err := sendDataToGCM(s, md, agg.sum, now, "", "", sumLabels...); err != nil {
return err
}
countLabels := append(labels, "count")
if err := sendDataToGCM(s, md, agg.count(), now, "", "", countLabels...); err != nil {
return err
}
test.Pass(ctx, "%s: %s\n", strings.Join(extraLabelKeys, " "), agg)
return nil
}