/*

Command slive reports on SLAs of live trace data

*/
package main

import (
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"

	"code.justin.tv/common/chitin"
	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/pgtables"
	"github.com/spenczar/tdigest"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
)

var (
	serverHostPort = flag.String("server", "trace-api.prod.us-west2.justin.tv:11143", "The server hostport")
	service        = flag.String("svc", "code.justin.tv/web/web", "The service to analyze")
	sampling       = flag.Float64("sample", 0.1, "Sampling rate")
	duration       = flag.Duration("t", time.Second, "How long to capture data before returning a result")
	slow           = flag.Duration("slow", 0, "Query duration which prompts logging")
	imatch         = flag.String("imatch", "", "Report on queries that match the provided string (case-insensitive)")
)

const (
	compression = 100 // sane default for tdigest
)

func main() {
	flag.Parse()

	err := chitin.ExperimentalTraceProcessOptIn()
	if err != nil {
		log.Fatalf("chitin opt in err=%q", err)
	}

	conn, err := grpc.Dial(*serverHostPort, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("fail to dial: %v", err)
	}
	defer conn.Close()
	client := api.NewTraceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), *duration)

	sigch := make(chan os.Signal, 1)
	signal.Notify(sigch, syscall.SIGINT)
	go func() {
		for _ = range sigch {
			cancel()
		}
	}()

	var comps []*api.Comparison
	if *service != "" {
		comps = append(comps, &api.Comparison{
			ServiceName: &api.StringComparison{Value: *service},
		})
	}

	firehose, err := client.Firehose(ctx, &api.FirehoseRequest{
		Sampling: *sampling,
		Query: &api.Query{
			Comparisons: comps,
		},
	})
	if err != nil {
		log.Fatalf("%v.Firehose(_) = _, %v", client, err)
	}

	result := consumeFirehose(firehose)
	if err := dump(result, os.Stdout); err != nil {
		log.Fatalf("dump err=%q", err)
	}
}

func consumeFirehose(f api.Trace_FirehoseClient) *results {
	r := newResults()
loop:
	for {
		tx, err := f.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			switch grpc.Code(err) {
			case codes.Canceled:
				break loop
			case codes.DeadlineExceeded:
				break loop
			default:
				log.Fatal(err.Error())
			}
		}
		r.addTx(tx)
	}
	return r
}

// interesting stuff starts here
// TODO(rhys): get rid of all the boilerplate above

type results struct {
	match   func(*api.Call) bool
	digests map[key]*val
}

type key struct {
	svc    string
	route  string
	tables string
	dest   string
}

type val struct {
	digest tdigest.TDigest
	count  int64
	total  time.Duration
}

func newResults() *results {
	return &results{
		match: func(c *api.Call) bool {
			if c == nil {
				return false
			}
			if c.Svc == nil {
				return false
			}
			return c.Svc.Name == *service || *service == ""
		},
		digests: make(map[key]*val),
	}
}

func (r *results) addTx(tx *api.Transaction) {
	r.visitCall(tx.GetRoot())
}

func (r *results) visitCall(c *api.Call) {
	if c == nil {
		return
	}
	for _, sub := range c.Subcalls {
		r.visitCall(sub)
	}

	if r.match(c) {
		route := ""
		if hp := c.GetParams().GetHttp(); hp != nil {
			route = hp.Route
		}

		for _, sub := range c.Subcalls {
			sql := sub.GetParams().GetSql()
			if sql == nil {
				continue
			}

			if *imatch != "" && !strings.Contains(strings.ToLower(sql.StrippedQuery), strings.ToLower(*imatch)) {
				continue
			}

			tables, err := pgtables.Cache.Tables(sql.StrippedQuery)
			if err != nil {
				tables = nil
			}
			var qt []string
			for _, t := range tables {
				qt = append(qt, strconv.Quote(t))
			}
			quoted := strings.Join(qt, ",")
			if tables == nil {
				quoted = "unknown"
			}

			svc := ""
			if c.Svc != nil {
				svc = c.Svc.Name
			}

			k := key{
				svc:    svc,
				route:  route,
				tables: quoted,
				dest:   sub.RequestSentTo,
			}
			v, ok := r.digests[k]
			if !ok {
				v = &val{
					digest: tdigest.New(compression),
				}
				r.digests[k] = v
			}
			d := api.ClientDuration(sub)
			v.digest.Add(d.Seconds(), 1)
			v.count += 1
			v.total += d

			if *slow > 0 && d >= *slow {
				fmt.Printf("%s d=%0.6fs dest=%q tables=%s query %s\n",
					api.ClientStart(sub).Truncate(time.Second).Format(time.RFC3339),
					d.Seconds(), sub.RequestSentTo, quoted, sql.StrippedQuery)
			}
		}
	}
}

func dump(r *results, w io.Writer) error {
	fmt.Fprintf(w, "digests=%d\n", len(r.digests))
	for _, k := range digestMap(r.digests).byTotal() {
		v := r.digests[k]
		fmt.Fprintf(w, "count=%-6d "+
			"p90=%0.6fs p99=%0.6fs "+
			"total=%07.3fs dest=%q svcname=%q rpc=%q tables=%s\n",
			v.count,
			v.digest.Quantile(0.9), v.digest.Quantile(0.99),
			v.total.Seconds(), k.dest, k.svc, k.route, k.tables)
	}
	return nil
}

type digestMap map[key]*val

func (m digestMap) byTotal() []key {
	var ks []key
	for k := range m {
		ks = append(ks, k)
	}
	sort.Sort(keys{
		ks: ks,
		less: func(i, j int) bool {
			ki, kj := ks[i], ks[j]
			vi, vj := m[ki], m[kj]
			return vi.total < vj.total
		},
	})
	return ks
}

func (m digestMap) byCount() []key {
	var ks []key
	for k := range m {
		ks = append(ks, k)
	}
	sort.Sort(keys{
		ks: ks,
		less: func(i, j int) bool {
			ki, kj := ks[i], ks[j]
			vi, vj := m[ki], m[kj]
			return vi.count < vj.count
		},
	})
	return ks
}

type keys struct {
	ks   []key
	less func(i, j int) bool
}

func (s keys) Len() int           { return len(s.ks) }
func (s keys) Swap(i, j int)      { s.ks[i], s.ks[j] = s.ks[j], s.ks[i] }
func (s keys) Less(i, j int) bool { return s.less(i, j) }
