oncall/serve.go - release.go.x.devtools - Git at Google

 // 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 (
 	"bytes"
 	"encoding/json"
 	"fmt"
 	"html"
 	"io/ioutil"
 	"math"
 	"net/http"
 	"sort"
 	"strconv"
 	"strings"
 	"time"

 	cloudmonitoring "google.golang.org/api/monitoring/v3"

 	"v.io/jiri"
 	"v.io/jiri/collect"
 	"v.io/jiri/tool"
 	"v.io/x/devtools/internal/cache"
 	"v.io/x/devtools/internal/monitoring"
 	"v.io/x/lib/cmdline"
 	"v.io/x/lib/gcm"
 )

 const (
 	numWorkers                = 64
 	resultTypeServiceLatency  = "resultTypeServiceLatency"
 	resultTypeServiceQPS      = "resultTypeServiceQPS"
 	resultTypeServiceCounters = "resultTypeServiceCounters"
 	resultTypeServiceMetadata = "resultTypeServiceMetadata"

 	getKubeDataTaskTypePod  = "getKubeDataTaskTypePod"
 	getKubeDataTaskTypeNode = "getKubeDataTaskTypeNode"

 	logsStyle = "font-family: monospace; font-size: 12px; white-space:pre-wrap; word-wrap: break-word;"
 )

 var (
 	addressFlag   string
 	cacheFlag     string
 	keyFileFlag   string
 	staticDirFlag string
 )

 type podSpec struct {
 	Metadata struct {
 		Name   string `json:"name"`
 		Uid    string `json:"uid"`
 		Labels struct {
 			Service string `json:"service"`
 			Version string `json:"version"`
 		} `json:"labels"`
 	} `json:"metadata"`
 	Spec struct {
 		NodeName   string `json:"nodeName"`
 		Containers []struct {
 			Name string `json:"name"`
 		} `json:"containers"`
 	} `json:"spec"`
 	Status struct {
 		Phase     string `json:"phase"`
 		StartTime string `json:"startTime"`
 	} `json:"status"`
 	zone    string
 	project string
 }

 type nodeSpec struct {
 	Metadata struct {
 		Name string `json:"name"`
 	} `json:"metadata"`
 	Spec struct {
 		ExternalID string `json:"externalID"`
 	} `json:"spec"`
 }

 type clusterLocation struct {
 	project string
 	zone    string
 }

 // Task and result for getKubeData.
 type getKubeDataTask struct {
 	location clusterLocation
 	taskType string
 }
 type getKubeDataResult struct {
 	pods     []*podSpec
 	nodes    []*nodeSpec
 	taskType string
 	err      error
 }

 // Task and result for getMetricWorker.
 type getMetricTask struct {
 	resultType  string
 	md          *cloudmonitoring.MetricDescriptor
 	metricName  string
 	pod         *podSpec
 	extraLabels map[string]string
 }
 type getMetricResult struct {
 	ResultType        string
 	Instance          string
 	Zone              string
 	Project           string
 	MetricName        string
 	ExtraLabels       map[string]string
 	CurrentValue      float64
 	MinValue          float64
 	MaxValue          float64
 	HistoryTimestamps []int64
 	HistoryValues     []float64
 	ErrMsg            string
 	PodName           string
 	PodUID            string
 	PodNode           string
 	PodStatus         string
 	MainContainer     string
 	ServiceVersion    string
 }

 type getMetricResults []getMetricResult

 func (m getMetricResults) Len() int           { return len(m) }
 func (m getMetricResults) Less(i, j int) bool { return m[i].Instance < m[j].Instance }
 func (m getMetricResults) Swap(i, j int)      { m[i], m[j] = m[j], m[i] }
 func (m getMetricResults) Sort()              { sort.Sort(m) }

 // Final result for getData endpoint.
 type getDataResult struct {
 	// These fields are indexed by metric names.
 	ServiceLatency  map[string]getMetricResults
 	ServiceQPS      map[string]getMetricResults
 	ServiceCounters map[string]getMetricResults
 	ServiceMetadata map[string]getMetricResults

 	Instances map[string]string // instances -> external ids
 	Oncalls   []string
 	MinTime   int64
 	MaxTime   int64
 }

 // Current service locations.
 var locations = []clusterLocation{
 	{
 		project: "vanadium-production",
 		zone:    "us-central1-c",
 	},
 	{
 		project: "vanadium-production",
 		zone:    "us-east1-c",
 	},
 	{
 		project: "vanadium-auth-production",
 		zone:    "us-central1-c",
 	},
 }

 func init() {
 	cmdServe.Flags.StringVar(&addressFlag, "address", ":8000", "Listening address for the server.")
 	cmdServe.Flags.StringVar(&cacheFlag, "cache", "", "Directory to use for caching files.")
 	cmdServe.Flags.StringVar(&keyFileFlag, "key", "", "The path to the service account's JSON credentials file.")
 	cmdServe.Flags.StringVar(&staticDirFlag, "static", "", "Directory to use for serving static files.")
 }

 // cmdServe represents the 'serve' command of the oncall tool.
 var cmdServe = &cmdline.Command{
 	Runner: cmdline.RunnerFunc(runServe),
 	Name:   "serve",
 	Short:  "Serve oncall dashboard data from Google Storage",
 	Long:   "Serve oncall dashboard data from Google Storage.",
 }

 func runServe(env *cmdline.Env, _ []string) (e error) {
 	jirix, err := jiri.NewX(env)
 	if err != nil {
 		return err
 	}

 	// Set up the root/cache directory.
 	root := cacheFlag
 	if root == "" {
 		tmpDir, err := jirix.NewSeq().TempDir("", "")
 		if err != nil {
 			return err
 		}
 		defer collect.Error(func() error { return jirix.NewSeq().RemoveAll(tmpDir).Done() }, &e)
 		root = tmpDir
 	}

 	// Start server.
 	http.HandleFunc("/data", func(w http.ResponseWriter, r *http.Request) {
 		dataHandler(jirix, root, w, r)
 	})
 	http.HandleFunc("/logs", func(w http.ResponseWriter, r *http.Request) {
 		logsHandler(jirix, root, w, r)
 	})
 	http.HandleFunc("/cfg", func(w http.ResponseWriter, r *http.Request) {
 		cfgHandler(jirix, root, w, r)
 	})
 	http.HandleFunc("/pic", func(w http.ResponseWriter, r *http.Request) {
 		picHandler(jirix, root, w, r)
 	})
 	staticHandler := http.FileServer(http.Dir(staticDirFlag))
 	http.Handle("/", staticHandler)
 	if err := http.ListenAndServe(addressFlag, nil); err != nil {
 		return fmt.Errorf("ListenAndServe(%s) failed: %v", addressFlag, err)
 	}

 	return nil
 }

 func dataHandler(jirix *jiri.X, root string, w http.ResponseWriter, r *http.Request) {
 	s, err := gcm.Authenticate(keyFileFlag)
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}

 	// Get start and end timestamps.
 	if err := r.ParseForm(); err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	now := time.Now()
 	startTimestamp := now.Unix() - 3600
 	endTimestamp := now.Unix()
 	strStartTimestamp := r.Form.Get("st")
 	if strStartTimestamp != "" {
 		var err error
 		startTimestamp, err = strconv.ParseInt(strStartTimestamp, 10, 64)
 		if err != nil {
 			respondWithError(jirix, err, w)
 			return
 		}
 	}
 	strEndTimestamp := r.Form.Get("et")
 	if strEndTimestamp != "" {
 		var err error
 		endTimestamp, err = strconv.ParseInt(strEndTimestamp, 10, 64)
 		if err != nil {
 			respondWithError(jirix, err, w)
 			return
 		}
 	}

 	// Get currently running pods and nodes from vanadium production clusters.
 	podsByServices, nodes, err := getKubeData(jirix)
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}

 	// Create tasks of getting metrics from GCM.
 	allTasks := []getMetricTask{}
 	mdServiceLatency, err := gcm.GetMetric("service-latency", "vanadium-production")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	mdServiceQPS, err := gcm.GetMetric("service-qps-total", "vanadium-production")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	mdServiceCounters, err := gcm.GetMetric("service-counters", "vanadium-production")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	mdServiceMetadata, err := gcm.GetMetric("service-metadata", "vanadium-production")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	for serviceName, pods := range podsByServices {
 		for _, pod := range pods {
 			allTasks = append(allTasks,
 				// Metadata.
 				getMetricTask{
 					resultType: resultTypeServiceMetadata,
 					md:         mdServiceMetadata,
 					metricName: serviceName,
 					pod:        pod,
 					extraLabels: map[string]string{
 						"metadata_name": "build age",
 					},
 				},
 				// QPS.
 				getMetricTask{
 					resultType: resultTypeServiceQPS,
 					md:         mdServiceQPS,
 					metricName: serviceName,
 					pod:        pod,
 				})
 			// Latency.
 			if serviceName == monitoring.SNIdentity {
 				for _, n := range []string{monitoring.SNMacaroon, monitoring.SNBinaryDischarger} {
 					allTasks = append(allTasks,
 						getMetricTask{
 							resultType: resultTypeServiceLatency,
 							md:         mdServiceLatency,
 							metricName: n,
 							pod:        pod,
 						})
 				}
 			} else {
 				allTasks = append(allTasks,
 					getMetricTask{
 						resultType: resultTypeServiceLatency,
 						md:         mdServiceLatency,
 						metricName: serviceName,
 						pod:        pod,
 					})
 			}
 			// Counters.
 			if serviceName == monitoring.SNMounttable {
 				allTasks = append(allTasks,
 					getMetricTask{
 						resultType: resultTypeServiceCounters,
 						md:         mdServiceCounters,
 						metricName: monitoring.MNMounttableMountedServers,
 						pod:        pod,
 					},
 					getMetricTask{
 						resultType: resultTypeServiceCounters,
 						md:         mdServiceCounters,
 						metricName: monitoring.MNMounttableNodes,
 						pod:        pod,
 					},
 				)
 			}
 		}
 	}

 	// Send tasks to workers.
 	numTasks := len(allTasks)
 	tasks := make(chan getMetricTask, numTasks)
 	taskResults := make(chan getMetricResult, numTasks)
 	for i := 0; i < numWorkers; i++ {
 		go getMetricWorker(jirix, s, time.Unix(startTimestamp, 0), time.Unix(endTimestamp, 0), tasks, taskResults)
 	}
 	for _, task := range allTasks {
 		tasks <- task
 	}
 	close(tasks)

 	// Process results.
 	result := getDataResult{
 		ServiceLatency:  map[string]getMetricResults{},
 		ServiceQPS:      map[string]getMetricResults{},
 		ServiceCounters: map[string]getMetricResults{},
 		ServiceMetadata: map[string]getMetricResults{},
 	}
 	for i := 0; i < numTasks; i++ {
 		r := <-taskResults
 		n := r.MetricName
 		switch r.ResultType {
 		case resultTypeServiceLatency:
 			result.ServiceLatency[n] = append(result.ServiceLatency[n], r)
 		case resultTypeServiceQPS:
 			result.ServiceQPS[n] = append(result.ServiceQPS[n], r)
 		case resultTypeServiceCounters:
 			result.ServiceCounters[n] = append(result.ServiceCounters[n], r)
 		case resultTypeServiceMetadata:
 			result.ServiceMetadata[n] = append(result.ServiceMetadata[n], r)
 		}
 	}
 	// Sort metrics by instance names.
 	fnSortMetrics := func(m map[string]getMetricResults) {
 		for n := range m {
 			m[n].Sort()
 		}
 	}
 	fnSortMetrics(result.ServiceLatency)
 	fnSortMetrics(result.ServiceQPS)
 	fnSortMetrics(result.ServiceCounters)
 	fnSortMetrics(result.ServiceMetadata)

 	result.MinTime = startTimestamp
 	result.MaxTime = endTimestamp
 	result.Instances = nodes

 	// Get oncalls.
 	oncalls, err := getOncalls(jirix)
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	result.Oncalls = oncalls

 	// Convert results to json and return it.
 	b, err := json.MarshalIndent(&result, "", "  ")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	w.Header().Set("Content-Type", "application/json")
 	w.Write(b)
 }

 func dataHandler2(jirix *jiri.X, root string, w http.ResponseWriter, r *http.Request) {
 	b, _ := ioutil.ReadFile("/usr/local/google/home/jingjin/data")
 	w.Header().Set("Content-Type", "application/json")
 	w.Write(b)
 }

 func getMetricWorker(jirix *jiri.X, s *cloudmonitoring.Service, startTime, endTime time.Time, tasks <-chan getMetricTask, results chan<- getMetricResult) {
 	for task := range tasks {
 		result := getMetricResult{
 			ResultType:     task.resultType,
 			Instance:       task.pod.Metadata.Name,
 			Zone:           task.pod.zone,
 			Project:        task.pod.project,
 			MetricName:     task.metricName,
 			ExtraLabels:    task.extraLabels,
 			CurrentValue:   -1,
 			MinValue:       math.MaxFloat64,
 			MaxValue:       0,
 			PodName:        task.pod.Metadata.Name,
 			PodUID:         task.pod.Metadata.Uid,
 			PodNode:        task.pod.Spec.NodeName,
 			PodStatus:      task.pod.Status.Phase,
 			MainContainer:  task.pod.Spec.Containers[0].Name,
 			ServiceVersion: task.pod.Metadata.Labels.Version,
 		}
 		filters := []string{
 			fmt.Sprintf("metric.type=%q", task.md.Type),
 			fmt.Sprintf("metric.label.metric_name=%q", task.metricName),
 			fmt.Sprintf("metric.label.gce_instance=%q", task.pod.Metadata.Name),
 			fmt.Sprintf("metric.label.gce_zone=%q", task.pod.zone),
 		}
 		for labelKey, labelValue := range task.extraLabels {
 			filters = append(filters, fmt.Sprintf("metric.label.%s=%q", labelKey, labelValue))
 		}
 		nextPageToken := ""
 		points := []*cloudmonitoring.Point{}
 		timestamps := []int64{}
 		values := []float64{}
 		for {
 			resp, err := s.Projects.TimeSeries.List("projects/vanadium-production").
 				IntervalStartTime(startTime.UTC().Format(time.RFC3339)).
 				IntervalEndTime(endTime.UTC().Format(time.RFC3339)).
 				Filter(strings.Join(filters, " AND ")).
 				PageToken(nextPageToken).Do()
 			if err != nil {
 				result.ErrMsg = fmt.Sprintf("List() failed: %v", err)
 				break
 			}
 			if len(resp.TimeSeries) > 0 {
 				// We should only get one timeseries.
 				ts := resp.TimeSeries[0]
 				points = append(points, ts.Points...)
 			}
 			if result.ErrMsg != "" {
 				break
 			}
 			nextPageToken = resp.NextPageToken
 			if nextPageToken == "" {
 				break
 			}
 		}
 		for i := len(points) - 1; i >= 0; i-- {
 			pt := points[i]
 			epochTime, err := time.Parse(time.RFC3339, pt.Interval.EndTime)
 			if err != nil {
 				result.ErrMsg = fmt.Sprintf("Parse(%s) failed: %v", pt.Interval.EndTime)
 				break
 			}
 			timestamp := epochTime.Unix()
 			timestamps = append(timestamps, timestamp)
 			value := pt.Value.DoubleValue
 			values = append(values, value)
 			result.MaxValue = math.Max(result.MaxValue, value)
 			result.MinValue = math.Min(result.MinValue, value)
 		}
 		result.HistoryTimestamps = timestamps
 		result.HistoryValues = values
 		if len(values) > 0 {
 			result.CurrentValue = values[len(values)-1]
 		}
 		results <- result
 	}
 }

 // getKubeData gets:
 // - pod data indexed by service names.
 // - node external ids indexed by names.
 func getKubeData(jirix *jiri.X) (map[string][]*podSpec, map[string]string, error) {
 	retPods := map[string][]*podSpec{}
 	retNodes := map[string]string{}

 	// Get pods data.
 	numTasks := len(locations) * 2
 	tasks := make(chan getKubeDataTask, numTasks)
 	taskResults := make(chan getKubeDataResult, numTasks)
 	for i := 0; i < numTasks; i++ {
 		go getKubeDataWorker(jirix, tasks, taskResults)
 	}
 	for _, loc := range locations {
 		tasks <- getKubeDataTask{
 			location: loc,
 			taskType: getKubeDataTaskTypePod,
 		}
 		tasks <- getKubeDataTask{
 			location: loc,
 			taskType: getKubeDataTaskTypeNode,
 		}
 	}
 	close(tasks)

 	pods := []*podSpec{}
 	nodes := []*nodeSpec{}
 	for i := 0; i < numTasks; i++ {
 		result := <-taskResults
 		switch result.taskType {
 		case getKubeDataTaskTypePod:
 			pods = append(pods, result.pods...)
 		case getKubeDataTaskTypeNode:
 			nodes = append(nodes, result.nodes...)
 		}
 	}

 	// Index pods data by service name.
 	for _, pod := range pods {
 		switch pod.Metadata.Labels.Service {
 		case "auth":
 			retPods[monitoring.SNIdentity] = append(retPods[monitoring.SNIdentity], pod)
 		case "benchmarks":
 			retPods[monitoring.SNBenchmark] = append(retPods[monitoring.SNBenchmark], pod)
 		case "mounttable":
 			retPods[monitoring.SNMounttable] = append(retPods[monitoring.SNMounttable], pod)
 		case "proxy":
 			retPods[monitoring.SNProxy] = append(retPods[monitoring.SNProxy], pod)
 		case "role":
 			retPods[monitoring.SNRole] = append(retPods[monitoring.SNRole], pod)
 		case "sb-allocator":
 			retPods[monitoring.SNAllocator] = append(retPods[monitoring.SNAllocator], pod)
 		}
 	}
 	// Index nodes names by ids.
 	for _, node := range nodes {
 		retNodes[node.Metadata.Name] = node.Spec.ExternalID
 	}

 	return retPods, retNodes, nil
 }

 func getKubeDataWorker(jirix *jiri.X, tasks <-chan getKubeDataTask, results chan<- getKubeDataResult) {
 	for task := range tasks {
 		kubeCtlArgs := []string{"get", "pods", "-o=json"}
 		if task.taskType == getKubeDataTaskTypeNode {
 			kubeCtlArgs = []string{"get", "nodes", "-o=json"}
 		}
 		var podItems struct {
 			Items []*podSpec `json:"items"`
 		}
 		var nodeItems struct {
 			Items []*nodeSpec `json:"items"`
 		}
 		out, err := runKubeCtl(jirix, task.location, kubeCtlArgs)
 		if err != nil {
 			results <- getKubeDataResult{
 				err: err,
 			}
 			continue
 		}
 		switch task.taskType {
 		case getKubeDataTaskTypePod:
 			if err := json.Unmarshal(out, &podItems); err != nil {
 				results <- getKubeDataResult{
 					err: fmt.Errorf("Unmarshal() failed: %v", err),
 				}
 				continue
 			}
 			for _, item := range podItems.Items {
 				item.zone = task.location.zone
 				item.project = task.location.project
 			}
 			results <- getKubeDataResult{
 				taskType: getKubeDataTaskTypePod,
 				pods:     podItems.Items,
 			}
 		case getKubeDataTaskTypeNode:
 			if err := json.Unmarshal(out, &nodeItems); err != nil {
 				results <- getKubeDataResult{
 					err: fmt.Errorf("Unmarshal() failed: %v", err),
 				}
 				continue
 			}
 			results <- getKubeDataResult{
 				taskType: getKubeDataTaskTypeNode,
 				nodes:    nodeItems.Items,
 			}
 		}
 	}
 }

 func runKubeCtl(jirix *jiri.X, location clusterLocation, kubeCtlArgs []string) ([]byte, error) {
 	f, err := ioutil.TempFile("", "")
 	if err != nil {
 		return nil, nil
 	}
 	defer jirix.NewSeq().RemoveAll(f.Name())
 	s := tool.NewContext(tool.ContextOpts{
 		Env: map[string]string{
 			"KUBECONFIG": f.Name(),
 		},
 	}).NewSeq()
 	var out bytes.Buffer
 	getCredsArgs := []string{
 		"container",
 		"clusters",
 		"get-credentials",
 		"vanadium",
 		"--project",
 		location.project,
 		"--zone",
 		location.zone,
 	}
 	if err := s.Run("gcloud", getCredsArgs...).Capture(&out, &out).Last("kubectl", kubeCtlArgs...); err != nil {
 		return nil, err
 	}
 	return out.Bytes(), nil
 }

 func getOncalls(jirix *jiri.X) ([]string, error) {
 	s := jirix.NewSeq()
 	var out bytes.Buffer
 	if err := s.Capture(&out, nil).Last("jiri", "oncall"); err != nil {
 		return nil, err
 	}
 	return strings.Split(strings.TrimSpace(out.String()), ","), nil
 }

 func logsHandler(jirix *jiri.X, root string, w http.ResponseWriter, r *http.Request) {
 	// Parse project, zone, pod name, and container.
 	f, err := parseForm(r, "p", "z", "d", "c")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	project, zone, pod, container := f["p"], f["z"], f["d"], f["c"]
 	out, err := runKubeCtl(jirix, clusterLocation{
 		project: project,
 		zone:    zone,
 	}, []string{"logs", pod, container})
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}

 	w.Header().Set("Content-Type", "text/html")
 	content := fmt.Sprintf("<pre style='%s'>%s</pre>", logsStyle, html.EscapeString(string(out)))
 	w.Write([]byte(content))
 }

 func cfgHandler(jirix *jiri.X, root string, w http.ResponseWriter, r *http.Request) {
 	// Parse project, zone, and pod name.
 	f, err := parseForm(r, "p", "z", "d")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	project, zone, pod := f["p"], f["z"], f["d"]
 	out, err := runKubeCtl(jirix, clusterLocation{
 		project: project,
 		zone:    zone,
 	}, []string{"get", "pods", pod, "-o=json"})
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}

 	w.Header().Set("Content-Type", "text/html")
 	content := fmt.Sprintf("<pre style='%s'>%s</pre>", logsStyle, html.EscapeString(string(out)))
 	w.Write([]byte(content))
 }

 func picHandler(jirix *jiri.X, root string, w http.ResponseWriter, r *http.Request) {
 	// Parameter "id" specifies the id of the pic.
 	f, err := parseForm(r, "id")
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	id := f["id"]

 	// Read picture file from Google Storage.
 	cachedFile, err := cache.StoreGoogleStorageFile(jirix, root, bucketPics, id+".png")
 	if err != nil {
 		// Read "_unknown.jpg" as fallback.
 		cachedFile, err = cache.StoreGoogleStorageFile(jirix, root, bucketPics, "_unknown.jpg")
 		if err != nil {
 			respondWithError(jirix, err, w)
 			return
 		}
 	}
 	bytes, err := jirix.NewSeq().ReadFile(cachedFile)
 	if err != nil {
 		respondWithError(jirix, err, w)
 		return
 	}
 	w.Header().Set("Content-Type", "image/jpeg")
 	w.Header().Set("Cache-control", "public, max-age=2592000")
 	w.Write(bytes)
 }

 func parseForm(r *http.Request, fields ...string) (map[string]string, error) {
 	m := map[string]string{}
 	r.ParseForm()
 	for _, f := range fields {
 		value := r.Form.Get(f)
 		if value == "" {
 			return nil, fmt.Errorf("parameter %q not found", f)
 		}
 		m[f] = value
 	}
 	return m, nil
 }

 func respondWithError(jirix *jiri.X, err error, w http.ResponseWriter) {
 	fmt.Fprintf(jirix.Stderr(), "%v\n", err)
 	http.Error(w, fmt.Sprintf("500 internal server error\n\n%v", err), http.StatusInternalServerError)
 }
	// 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 (
	"bytes"
	"encoding/json"
	"fmt"
	"html"
	"io/ioutil"
	"math"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	cloudmonitoring "google.golang.org/api/monitoring/v3"

	"v.io/jiri"
	"v.io/jiri/collect"
	"v.io/jiri/tool"
	"v.io/x/devtools/internal/cache"
	"v.io/x/devtools/internal/monitoring"
	"v.io/x/lib/cmdline"
	"v.io/x/lib/gcm"
	)

	const (
	numWorkers = 64
	resultTypeServiceLatency = "resultTypeServiceLatency"
	resultTypeServiceQPS = "resultTypeServiceQPS"
	resultTypeServiceCounters = "resultTypeServiceCounters"
	resultTypeServiceMetadata = "resultTypeServiceMetadata"

	getKubeDataTaskTypePod = "getKubeDataTaskTypePod"
	getKubeDataTaskTypeNode = "getKubeDataTaskTypeNode"

	logsStyle = "font-family: monospace; font-size: 12px; white-space:pre-wrap; word-wrap: break-word;"
	)

	var (
	addressFlag string
	cacheFlag string
	keyFileFlag string
	staticDirFlag string
	)

	type podSpec struct {
	Metadata struct {
	Name string `json:"name"`
	Uid string `json:"uid"`
	Labels struct {
	Service string `json:"service"`
	Version string `json:"version"`
	} `json:"labels"`
	} `json:"metadata"`
	Spec struct {
	NodeName string `json:"nodeName"`
	Containers []struct {
	Name string `json:"name"`
	} `json:"containers"`
	} `json:"spec"`
	Status struct {
	Phase string `json:"phase"`
	StartTime string `json:"startTime"`
	} `json:"status"`
	zone string
	project string
	}

	type nodeSpec struct {
	Metadata struct {
	Name string `json:"name"`
	} `json:"metadata"`
	Spec struct {
	ExternalID string `json:"externalID"`
	} `json:"spec"`
	}

	type clusterLocation struct {
	project string
	zone string
	}

	// Task and result for getKubeData.
	type getKubeDataTask struct {
	location clusterLocation
	taskType string
	}
	type getKubeDataResult struct {
	pods []*podSpec
	nodes []*nodeSpec
	taskType string
	err error
	}

	// Task and result for getMetricWorker.
	type getMetricTask struct {
	resultType string
	md *cloudmonitoring.MetricDescriptor
	metricName string
	pod *podSpec
	extraLabels map[string]string
	}
	type getMetricResult struct {
	ResultType string
	Instance string
	Zone string
	Project string
	MetricName string
	ExtraLabels map[string]string
	CurrentValue float64
	MinValue float64
	MaxValue float64
	HistoryTimestamps []int64
	HistoryValues []float64
	ErrMsg string
	PodName string
	PodUID string
	PodNode string
	PodStatus string
	MainContainer string
	ServiceVersion string
	}

	type getMetricResults []getMetricResult

	func (m getMetricResults) Len() int { return len(m) }
	func (m getMetricResults) Less(i, j int) bool { return m[i].Instance < m[j].Instance }
	func (m getMetricResults) Swap(i, j int) { m[i], m[j] = m[j], m[i] }
	func (m getMetricResults) Sort() { sort.Sort(m) }

	// Final result for getData endpoint.
	type getDataResult struct {
	// These fields are indexed by metric names.
	ServiceLatency map[string]getMetricResults
	ServiceQPS map[string]getMetricResults
	ServiceCounters map[string]getMetricResults
	ServiceMetadata map[string]getMetricResults

	Instances map[string]string // instances -> external ids
	Oncalls []string
	MinTime int64
	MaxTime int64
	}

	// Current service locations.
	var locations = []clusterLocation{
	{
	project: "vanadium-production",
	zone: "us-central1-c",
	},
	{
	project: "vanadium-production",
	zone: "us-east1-c",
	},
	{
	project: "vanadium-auth-production",
	zone: "us-central1-c",
	},
	}

	func init() {
	cmdServe.Flags.StringVar(&addressFlag, "address", ":8000", "Listening address for the server.")
	cmdServe.Flags.StringVar(&cacheFlag, "cache", "", "Directory to use for caching files.")
	cmdServe.Flags.StringVar(&keyFileFlag, "key", "", "The path to the service account's JSON credentials file.")
	cmdServe.Flags.StringVar(&staticDirFlag, "static", "", "Directory to use for serving static files.")
	}

	// cmdServe represents the 'serve' command of the oncall tool.
	var cmdServe = &cmdline.Command{
	Runner: cmdline.RunnerFunc(runServe),
	Name: "serve",
	Short: "Serve oncall dashboard data from Google Storage",
	Long: "Serve oncall dashboard data from Google Storage.",
	}

	func runServe(env *cmdline.Env, _ []string) (e error) {
	jirix, err := jiri.NewX(env)
	if err != nil {
	return err
	}

	// Set up the root/cache directory.
	root := cacheFlag
	if root == "" {
	tmpDir, err := jirix.NewSeq().TempDir("", "")
	if err != nil {
	return err
	}
	defer collect.Error(func() error { return jirix.NewSeq().RemoveAll(tmpDir).Done() }, &e)
	root = tmpDir
	}

	// Start server.
	http.HandleFunc("/data", func(w http.ResponseWriter, r *http.Request) {
	dataHandler(jirix, root, w, r)
	})
	http.HandleFunc("/logs", func(w http.ResponseWriter, r *http.Request) {
	logsHandler(jirix, root, w, r)
	})
	http.HandleFunc("/cfg", func(w http.ResponseWriter, r *http.Request) {
	cfgHandler(jirix, root, w, r)
	})
	http.HandleFunc("/pic", func(w http.ResponseWriter, r *http.Request) {
	picHandler(jirix, root, w, r)
	})
	staticHandler := http.FileServer(http.Dir(staticDirFlag))
	http.Handle("/", staticHandler)
	if err := http.ListenAndServe(addressFlag, nil); err != nil {
	return fmt.Errorf("ListenAndServe(%s) failed: %v", addressFlag, err)
	}

	return nil
	}

	func dataHandler(jirix jiri.X, root string, w http.ResponseWriter, r http.Request) {
	s, err := gcm.Authenticate(keyFileFlag)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}

	// Get start and end timestamps.
	if err := r.ParseForm(); err != nil {
	respondWithError(jirix, err, w)
	return
	}
	now := time.Now()
	startTimestamp := now.Unix() - 3600
	endTimestamp := now.Unix()
	strStartTimestamp := r.Form.Get("st")
	if strStartTimestamp != "" {
	var err error
	startTimestamp, err = strconv.ParseInt(strStartTimestamp, 10, 64)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	}
	strEndTimestamp := r.Form.Get("et")
	if strEndTimestamp != "" {
	var err error
	endTimestamp, err = strconv.ParseInt(strEndTimestamp, 10, 64)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	}

	// Get currently running pods and nodes from vanadium production clusters.
	podsByServices, nodes, err := getKubeData(jirix)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}

	// Create tasks of getting metrics from GCM.
	allTasks := []getMetricTask{}
	mdServiceLatency, err := gcm.GetMetric("service-latency", "vanadium-production")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	mdServiceQPS, err := gcm.GetMetric("service-qps-total", "vanadium-production")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	mdServiceCounters, err := gcm.GetMetric("service-counters", "vanadium-production")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	mdServiceMetadata, err := gcm.GetMetric("service-metadata", "vanadium-production")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	for serviceName, pods := range podsByServices {
	for _, pod := range pods {
	allTasks = append(allTasks,
	// Metadata.
	getMetricTask{
	resultType: resultTypeServiceMetadata,
	md: mdServiceMetadata,
	metricName: serviceName,
	pod: pod,
	extraLabels: map[string]string{
	"metadata_name": "build age",
	},
	},
	// QPS.
	getMetricTask{
	resultType: resultTypeServiceQPS,
	md: mdServiceQPS,
	metricName: serviceName,
	pod: pod,
	})
	// Latency.
	if serviceName == monitoring.SNIdentity {
	for _, n := range []string{monitoring.SNMacaroon, monitoring.SNBinaryDischarger} {
	allTasks = append(allTasks,
	getMetricTask{
	resultType: resultTypeServiceLatency,
	md: mdServiceLatency,
	metricName: n,
	pod: pod,
	})
	}
	} else {
	allTasks = append(allTasks,
	getMetricTask{
	resultType: resultTypeServiceLatency,
	md: mdServiceLatency,
	metricName: serviceName,
	pod: pod,
	})
	}
	// Counters.
	if serviceName == monitoring.SNMounttable {
	allTasks = append(allTasks,
	getMetricTask{
	resultType: resultTypeServiceCounters,
	md: mdServiceCounters,
	metricName: monitoring.MNMounttableMountedServers,
	pod: pod,
	},
	getMetricTask{
	resultType: resultTypeServiceCounters,
	md: mdServiceCounters,
	metricName: monitoring.MNMounttableNodes,
	pod: pod,
	},
	)
	}
	}
	}

	// Send tasks to workers.
	numTasks := len(allTasks)
	tasks := make(chan getMetricTask, numTasks)
	taskResults := make(chan getMetricResult, numTasks)
	for i := 0; i < numWorkers; i++ {
	go getMetricWorker(jirix, s, time.Unix(startTimestamp, 0), time.Unix(endTimestamp, 0), tasks, taskResults)
	}
	for _, task := range allTasks {
	tasks <- task
	}
	close(tasks)

	// Process results.
	result := getDataResult{
	ServiceLatency: map[string]getMetricResults{},
	ServiceQPS: map[string]getMetricResults{},
	ServiceCounters: map[string]getMetricResults{},
	ServiceMetadata: map[string]getMetricResults{},
	}
	for i := 0; i < numTasks; i++ {
	r := <-taskResults
	n := r.MetricName
	switch r.ResultType {
	case resultTypeServiceLatency:
	result.ServiceLatency[n] = append(result.ServiceLatency[n], r)
	case resultTypeServiceQPS:
	result.ServiceQPS[n] = append(result.ServiceQPS[n], r)
	case resultTypeServiceCounters:
	result.ServiceCounters[n] = append(result.ServiceCounters[n], r)
	case resultTypeServiceMetadata:
	result.ServiceMetadata[n] = append(result.ServiceMetadata[n], r)
	}
	}
	// Sort metrics by instance names.
	fnSortMetrics := func(m map[string]getMetricResults) {
	for n := range m {
	m[n].Sort()
	}
	}
	fnSortMetrics(result.ServiceLatency)
	fnSortMetrics(result.ServiceQPS)
	fnSortMetrics(result.ServiceCounters)
	fnSortMetrics(result.ServiceMetadata)

	result.MinTime = startTimestamp
	result.MaxTime = endTimestamp
	result.Instances = nodes

	// Get oncalls.
	oncalls, err := getOncalls(jirix)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	result.Oncalls = oncalls

	// Convert results to json and return it.
	b, err := json.MarshalIndent(&result, "", " ")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	w.Header().Set("Content-Type", "application/json")
	w.Write(b)
	}

	func dataHandler2(jirix jiri.X, root string, w http.ResponseWriter, r http.Request) {
	b, _ := ioutil.ReadFile("/usr/local/google/home/jingjin/data")
	w.Header().Set("Content-Type", "application/json")
	w.Write(b)
	}

	func getMetricWorker(jirix jiri.X, s cloudmonitoring.Service, startTime, endTime time.Time, tasks <-chan getMetricTask, results chan<- getMetricResult) {
	for task := range tasks {
	result := getMetricResult{
	ResultType: task.resultType,
	Instance: task.pod.Metadata.Name,
	Zone: task.pod.zone,
	Project: task.pod.project,
	MetricName: task.metricName,
	ExtraLabels: task.extraLabels,
	CurrentValue: -1,
	MinValue: math.MaxFloat64,
	MaxValue: 0,
	PodName: task.pod.Metadata.Name,
	PodUID: task.pod.Metadata.Uid,
	PodNode: task.pod.Spec.NodeName,
	PodStatus: task.pod.Status.Phase,
	MainContainer: task.pod.Spec.Containers[0].Name,
	ServiceVersion: task.pod.Metadata.Labels.Version,
	}
	filters := []string{
	fmt.Sprintf("metric.type=%q", task.md.Type),
	fmt.Sprintf("metric.label.metric_name=%q", task.metricName),
	fmt.Sprintf("metric.label.gce_instance=%q", task.pod.Metadata.Name),
	fmt.Sprintf("metric.label.gce_zone=%q", task.pod.zone),
	}
	for labelKey, labelValue := range task.extraLabels {
	filters = append(filters, fmt.Sprintf("metric.label.%s=%q", labelKey, labelValue))
	}
	nextPageToken := ""
	points := []*cloudmonitoring.Point{}
	timestamps := []int64{}
	values := []float64{}
	for {
	resp, err := s.Projects.TimeSeries.List("projects/vanadium-production").
	IntervalStartTime(startTime.UTC().Format(time.RFC3339)).
	IntervalEndTime(endTime.UTC().Format(time.RFC3339)).
	Filter(strings.Join(filters, " AND ")).
	PageToken(nextPageToken).Do()
	if err != nil {
	result.ErrMsg = fmt.Sprintf("List() failed: %v", err)
	break
	}
	if len(resp.TimeSeries) > 0 {
	// We should only get one timeseries.
	ts := resp.TimeSeries[0]
	points = append(points, ts.Points...)
	}
	if result.ErrMsg != "" {
	break
	}
	nextPageToken = resp.NextPageToken
	if nextPageToken == "" {
	break
	}
	}
	for i := len(points) - 1; i >= 0; i-- {
	pt := points[i]
	epochTime, err := time.Parse(time.RFC3339, pt.Interval.EndTime)
	if err != nil {
	result.ErrMsg = fmt.Sprintf("Parse(%s) failed: %v", pt.Interval.EndTime)
	break
	}
	timestamp := epochTime.Unix()
	timestamps = append(timestamps, timestamp)
	value := pt.Value.DoubleValue
	values = append(values, value)
	result.MaxValue = math.Max(result.MaxValue, value)
	result.MinValue = math.Min(result.MinValue, value)
	}
	result.HistoryTimestamps = timestamps
	result.HistoryValues = values
	if len(values) > 0 {
	result.CurrentValue = values[len(values)-1]
	}
	results <- result
	}
	}

	// getKubeData gets:
	// - pod data indexed by service names.
	// - node external ids indexed by names.
	func getKubeData(jirix jiri.X) (map[string][]podSpec, map[string]string, error) {
	retPods := map[string][]*podSpec{}
	retNodes := map[string]string{}

	// Get pods data.
	numTasks := len(locations) * 2
	tasks := make(chan getKubeDataTask, numTasks)
	taskResults := make(chan getKubeDataResult, numTasks)
	for i := 0; i < numTasks; i++ {
	go getKubeDataWorker(jirix, tasks, taskResults)
	}
	for _, loc := range locations {
	tasks <- getKubeDataTask{
	location: loc,
	taskType: getKubeDataTaskTypePod,
	}
	tasks <- getKubeDataTask{
	location: loc,
	taskType: getKubeDataTaskTypeNode,
	}
	}
	close(tasks)

	pods := []*podSpec{}
	nodes := []*nodeSpec{}
	for i := 0; i < numTasks; i++ {
	result := <-taskResults
	switch result.taskType {
	case getKubeDataTaskTypePod:
	pods = append(pods, result.pods...)
	case getKubeDataTaskTypeNode:
	nodes = append(nodes, result.nodes...)
	}
	}

	// Index pods data by service name.
	for _, pod := range pods {
	switch pod.Metadata.Labels.Service {
	case "auth":
	retPods[monitoring.SNIdentity] = append(retPods[monitoring.SNIdentity], pod)
	case "benchmarks":
	retPods[monitoring.SNBenchmark] = append(retPods[monitoring.SNBenchmark], pod)
	case "mounttable":
	retPods[monitoring.SNMounttable] = append(retPods[monitoring.SNMounttable], pod)
	case "proxy":
	retPods[monitoring.SNProxy] = append(retPods[monitoring.SNProxy], pod)
	case "role":
	retPods[monitoring.SNRole] = append(retPods[monitoring.SNRole], pod)
	case "sb-allocator":
	retPods[monitoring.SNAllocator] = append(retPods[monitoring.SNAllocator], pod)
	}
	}
	// Index nodes names by ids.
	for _, node := range nodes {
	retNodes[node.Metadata.Name] = node.Spec.ExternalID
	}

	return retPods, retNodes, nil
	}

	func getKubeDataWorker(jirix *jiri.X, tasks <-chan getKubeDataTask, results chan<- getKubeDataResult) {
	for task := range tasks {
	kubeCtlArgs := []string{"get", "pods", "-o=json"}
	if task.taskType == getKubeDataTaskTypeNode {
	kubeCtlArgs = []string{"get", "nodes", "-o=json"}
	}
	var podItems struct {
	Items []*podSpec `json:"items"`
	}
	var nodeItems struct {
	Items []*nodeSpec `json:"items"`
	}
	out, err := runKubeCtl(jirix, task.location, kubeCtlArgs)
	if err != nil {
	results <- getKubeDataResult{
	err: err,
	}
	continue
	}
	switch task.taskType {
	case getKubeDataTaskTypePod:
	if err := json.Unmarshal(out, &podItems); err != nil {
	results <- getKubeDataResult{
	err: fmt.Errorf("Unmarshal() failed: %v", err),
	}
	continue
	}
	for _, item := range podItems.Items {
	item.zone = task.location.zone
	item.project = task.location.project
	}
	results <- getKubeDataResult{
	taskType: getKubeDataTaskTypePod,
	pods: podItems.Items,
	}
	case getKubeDataTaskTypeNode:
	if err := json.Unmarshal(out, &nodeItems); err != nil {
	results <- getKubeDataResult{
	err: fmt.Errorf("Unmarshal() failed: %v", err),
	}
	continue
	}
	results <- getKubeDataResult{
	taskType: getKubeDataTaskTypeNode,
	nodes: nodeItems.Items,
	}
	}
	}
	}

	func runKubeCtl(jirix *jiri.X, location clusterLocation, kubeCtlArgs []string) ([]byte, error) {
	f, err := ioutil.TempFile("", "")
	if err != nil {
	return nil, nil
	}
	defer jirix.NewSeq().RemoveAll(f.Name())
	s := tool.NewContext(tool.ContextOpts{
	Env: map[string]string{
	"KUBECONFIG": f.Name(),
	},
	}).NewSeq()
	var out bytes.Buffer
	getCredsArgs := []string{
	"container",
	"clusters",
	"get-credentials",
	"vanadium",
	"--project",
	location.project,
	"--zone",
	location.zone,
	}
	if err := s.Run("gcloud", getCredsArgs...).Capture(&out, &out).Last("kubectl", kubeCtlArgs...); err != nil {
	return nil, err
	}
	return out.Bytes(), nil
	}

	func getOncalls(jirix *jiri.X) ([]string, error) {
	s := jirix.NewSeq()
	var out bytes.Buffer
	if err := s.Capture(&out, nil).Last("jiri", "oncall"); err != nil {
	return nil, err
	}
	return strings.Split(strings.TrimSpace(out.String()), ","), nil
	}

	func logsHandler(jirix jiri.X, root string, w http.ResponseWriter, r http.Request) {
	// Parse project, zone, pod name, and container.
	f, err := parseForm(r, "p", "z", "d", "c")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	project, zone, pod, container := f["p"], f["z"], f["d"], f["c"]
	out, err := runKubeCtl(jirix, clusterLocation{
	project: project,
	zone: zone,
	}, []string{"logs", pod, container})
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}

	w.Header().Set("Content-Type", "text/html")
	content := fmt.Sprintf("<pre style='%s'>%s</pre>", logsStyle, html.EscapeString(string(out)))
	w.Write([]byte(content))
	}

	func cfgHandler(jirix jiri.X, root string, w http.ResponseWriter, r http.Request) {
	// Parse project, zone, and pod name.
	f, err := parseForm(r, "p", "z", "d")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	project, zone, pod := f["p"], f["z"], f["d"]
	out, err := runKubeCtl(jirix, clusterLocation{
	project: project,
	zone: zone,
	}, []string{"get", "pods", pod, "-o=json"})
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}

	w.Header().Set("Content-Type", "text/html")
	content := fmt.Sprintf("<pre style='%s'>%s</pre>", logsStyle, html.EscapeString(string(out)))
	w.Write([]byte(content))
	}

	func picHandler(jirix jiri.X, root string, w http.ResponseWriter, r http.Request) {
	// Parameter "id" specifies the id of the pic.
	f, err := parseForm(r, "id")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	id := f["id"]

	// Read picture file from Google Storage.
	cachedFile, err := cache.StoreGoogleStorageFile(jirix, root, bucketPics, id+".png")
	if err != nil {
	// Read "_unknown.jpg" as fallback.
	cachedFile, err = cache.StoreGoogleStorageFile(jirix, root, bucketPics, "_unknown.jpg")
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	}
	bytes, err := jirix.NewSeq().ReadFile(cachedFile)
	if err != nil {
	respondWithError(jirix, err, w)
	return
	}
	w.Header().Set("Content-Type", "image/jpeg")
	w.Header().Set("Cache-control", "public, max-age=2592000")
	w.Write(bytes)
	}

	func parseForm(r *http.Request, fields ...string) (map[string]string, error) {
	m := map[string]string{}
	r.ParseForm()
	for _, f := range fields {
	value := r.Form.Get(f)
	if value == "" {
	return nil, fmt.Errorf("parameter %q not found", f)
	}
	m[f] = value
	}
	return m, nil
	}

	func respondWithError(jirix *jiri.X, err error, w http.ResponseWriter) {
	fmt.Fprintf(jirix.Stderr(), "%v\n", err)
	http.Error(w, fmt.Sprintf("500 internal server error\n\n%v", err), http.StatusInternalServerError)
	}