[grpc] add sybild grpc service (#37)

[proto] add proto file
[grpc] simplify response
[grpc] stub out implementation
[grpc] add sybild cmd skeleton
[grpc] add stub comments to satisfy linter
[grpc] use x/net/context for go<1.9
[grpc] add Trim methods
[grpc] add initial grpc implementation
[grpc] remove panics
[grpc] implement GetTable
[grpc] implement more of query
[grpc] query - implement samples
[grpc] populate histogram
[grpc] fixup
[grpc] use exec, implement more of query
[query] use x/net/context for 1.6 support
[grpc] implement time series queries
[grpc] set large fields to uint64
[grpc] add filter support
[grpc] populate distinct count
[grpc] add set fields support in samples
[grpc] add basic doc.go
[grpc] implement ReadIngestionLog
[grpc] implement ingest

Author: tmc

main.go

@@ -20,6 +20,7 @@ func setupCommands() {
 	CMD_FUNCS["inspect"] = cmd.RunInspectCmdLine
 	CMD_FUNCS["aggregate"] = cmd.RunAggregateCmdLine
 	CMD_FUNCS["version"] = cmd.RunVersionCmdLine
+	CMD_FUNCS["serve"] = cmd.RunServeCmdLine
 
 	for k, _ := range CMD_FUNCS {
 		CMD_KEYS = append(CMD_KEYS, k)

src/cmd/cmd_ingest.go

@@ -3,16 +3,25 @@ package sybil_cmd
 import sybil "github.com/logv/sybil/src/lib"
 
 import (
+	"bufio"
+	"bytes"
+
 	"encoding/csv"
 	"encoding/json"
 	"flag"
 	"fmt"
+	"github.com/golang/protobuf/jsonpb"
+	"github.com/golang/protobuf/ptypes/struct"
 	"io"
 	"os"
 	"strconv"
 	"strings"
 	"syscall"
 	"time"
+
+	pb "github.com/logv/sybil/src/sybilpb"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
 )
 
 type Dictionary map[string]interface{}
@@ -220,6 +229,20 @@ func RunIngestCmdLine() {
 	f_REOPEN := flag.String("infile", "", "input file to use (instead of stdin)")
 
 	flag.Parse()
+	if sybil.FLAGS.DIAL != "" {
+		var r io.ReadCloser = os.Stdin
+
+		var err error
+		if *f_REOPEN != "" {
+			r, err = os.Open(*f_REOPEN)
+		}
+		if err != nil {
+			return
+		}
+
+		defer r.Close()
+		runIngestGRPC(&sybil.FLAGS, r)
+	}
 
 	digestfile := fmt.Sprintf("%s", *ingestfile)
 
@@ -286,3 +309,58 @@ func RunIngestCmdLine() {
 
 	t.IngestRecords(digestfile)
 }
+
+func runIngestGRPC(flags *sybil.FlagDefs, r io.Reader) error {
+	ctx := context.Background()
+	opts := []grpc.DialOption{
+		// todo
+		grpc.WithInsecure(),
+	}
+	conn, err := grpc.Dial(flags.DIAL, opts...)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	c := pb.NewSybilClient(conn)
+	m := &jsonpb.Marshaler{OrigName: true}
+	maxErrs := 100
+	var errs int
+	var vals []*structpb.Struct
+	s := bufio.NewScanner(r)
+	for s.Scan() {
+		v := &structpb.Struct{}
+		if err := jsonpb.Unmarshal(bytes.NewReader(s.Bytes()), v); err != nil {
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				sybil.Debug("ERR", err)
+				errs++
+				if errs > maxErrs {
+					break
+				}
+				continue
+			}
+		}
+		vals = append(vals, v)
+	}
+	if err := s.Err(); err != nil {
+		return err
+	}
+	i := &pb.IngestRequest{
+		Dataset: flags.TABLE,
+		Records: vals,
+	}
+	qr, err := c.Ingest(ctx, i)
+	if err != nil {
+		return err
+	}
+	if err := m.Marshal(os.Stdout, qr); err != nil {
+		return err
+	}
+	if errs > 0 {
+		fmt.Fprintln(os.Stderr, "exiting due to error threshold being reached:", errs)
+		os.Exit(errs)
+	}
+	return nil
+}

src/cmd/cmd_query.go

@@ -4,11 +4,20 @@ import sybil "github.com/logv/sybil/src/lib"
 
 import "fmt"
 import "flag"
+import "os"
 import "strings"
 import "time"
 import "path"
 import "runtime/debug"
 
+import (
+	"golang.org/x/net/context"
+
+	"github.com/golang/protobuf/jsonpb"
+	pb "github.com/logv/sybil/src/sybilpb"
+	"google.golang.org/grpc"
+)
+
 var MAX_RECORDS_NO_GC = 4 * 1000 * 1000 // 4 million
 
 const (
@@ -68,6 +77,16 @@ func RunQueryCmdLine() {
 	addPrintFlags()
 	flag.Parse()
 
+	runQueryCmdLine()
+
+}
+
+func runQueryCmdLine() {
+	if sybil.FLAGS.DIAL != "" {
+		runQueryGRPC(&sybil.FLAGS)
+		return
+	}
+
 	if sybil.FLAGS.DECODE_FLAGS {
 		sybil.DecodeFlags()
 	}
@@ -301,3 +320,88 @@ func RunQueryCmdLine() {
 	}
 
 }
+
+func split(s, sep string) []string {
+	if s == "" {
+		return nil
+	}
+	return strings.Split(s, sep)
+}
+func splitFilters(s, sep, fSep string) []*pb.QueryFilter {
+	if s == "" {
+		return nil
+	}
+	var result []*pb.QueryFilter
+	for _, filter := range strings.Split(s, sep) {
+		parts := strings.Split(filter, fSep)
+		// TODO: check filter validity
+		result = append(result, &pb.QueryFilter{
+			Column: parts[0],
+			Op:     pb.QueryFilterOp(pb.QueryFilterOp_value[strings.ToUpper(parts[1])]),
+			Value:  parts[2],
+		})
+	}
+	return result
+}
+func runQueryGRPC(flags *sybil.FlagDefs) error {
+	ctx := context.Background()
+	opts := []grpc.DialOption{
+		// todo
+		grpc.WithInsecure(),
+	}
+	conn, err := grpc.Dial(flags.DIAL, opts...)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	c := pb.NewSybilClient(conn)
+	m := &jsonpb.Marshaler{OrigName: true}
+	if flags.LIST_TABLES {
+		r, err := c.ListTables(ctx, &pb.ListTablesRequest{})
+		if err != nil {
+			return err
+		}
+		return m.Marshal(os.Stdout, r)
+	}
+	if flags.PRINT_INFO {
+		r, err := c.GetTable(ctx, &pb.GetTableRequest{
+			Name: flags.TABLE,
+		})
+		if err != nil {
+			return err
+		}
+		return m.Marshal(os.Stdout, r)
+	}
+	fs, flts := flags.FIELD_SEPARATOR, flags.FILTER_SEPARATOR
+	q := &pb.QueryRequest{
+		Dataset:          flags.TABLE,
+		Ints:             split(flags.INTS, fs),
+		Strs:             split(flags.STRS, fs),
+		GroupBy:          split(flags.GROUPS, fs),
+		DistinctGroupBy:  split(flags.DISTINCT, fs),
+		Limit:            int64(flags.LIMIT),
+		SortBy:           flags.SORT,
+		IntFilters:       splitFilters(flags.INT_FILTERS, fs, flts),
+		StrFilters:       splitFilters(flags.STR_FILTERS, fs, flts),
+		SetFilters:       splitFilters(flags.SET_FILTERS, fs, flts),
+		ReadIngestionLog: flags.READ_ROWSTORE,
+		// TODO: replacements
+	}
+	if flags.SAMPLES {
+		q.Type = pb.QueryType_SAMPLES
+	}
+	if flags.OP == sybil.OP_AVG {
+		q.Op = pb.QueryOp_AVERAGE
+	} else if flags.OP == sybil.OP_HIST {
+		q.Op = pb.QueryOp_HISTOGRAM
+		q.Type = pb.QueryType_DISTRIBUTION
+	}
+	if flags.TIME {
+		q.Type = pb.QueryType_TIME_SERIES
+	}
+	qr, err := c.Query(ctx, q)
+	if err != nil {
+		return err
+	}
+	return m.Marshal(os.Stdout, qr)
+}

src/cmd/cmd_serve.go

@@ -0,0 +1,41 @@
+package sybil_cmd
+
+import (
+	"flag"
+	"fmt"
+	"net"
+	"os"
+
+	sybil "github.com/logv/sybil/src/lib"
+	"github.com/logv/sybil/src/sybild"
+	pb "github.com/logv/sybil/src/sybilpb"
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+)
+
+const defaultServeListenAddr = "localhost:7000"
+
+func RunServeCmdLine() {
+	flag.Parse()
+	if err := runServeCmdLine(&sybil.FLAGS); err != nil {
+		fmt.Fprintln(os.Stderr, errors.Wrap(err, "serve"))
+		os.Exit(1)
+	}
+}
+
+func runServeCmdLine(flags *sybil.FlagDefs) error {
+	// TODO: handle signals, shutdown
+	// TODO: auth, tls
+	// TODO: add configurable listening address
+	lis, err := net.Listen("tcp", defaultServeListenAddr)
+	if err != nil {
+		return err
+	}
+	s, err := sybild.NewServer()
+	if err != nil {
+		return err
+	}
+	grpcServer := grpc.NewServer()
+	pb.RegisterSybilServer(grpcServer, s)
+	return grpcServer.Serve(lis)
+}

src/cmd/sybild/main.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/logv/sybil/src/sybild"
+)
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintln(os.Stderr, "sybild:", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	s, err := sybild.NewServer()
+	if err != nil {
+		return err
+	}
+	_ = s
+	return nil
+}

src/lib/aggregate.go

@@ -22,10 +22,10 @@ type op string
 const SORT_COUNT = "$COUNT"
 
 const (
-	NO_OP       = iota
-	OP_AVG      = iota
-	OP_HIST     = iota
-	OP_DISTINCT = iota
+	NO_OP       = ""
+	OP_AVG      = "avg"
+	OP_HIST     = "hist"
+	OP_DISTINCT = "distinct"
 )
 
 var GROUP_DELIMITER = "\t"

src/lib/config.go

@@ -88,6 +88,8 @@ type FlagDefs struct {
 
 	UPDATE_TABLE_INFO bool
 	SKIP_OUTLIERS     bool
+
+	DIAL string // GRPC dialing
 }
 
 type StrReplace struct {
@@ -140,6 +142,8 @@ func setDefaults() {
 	flag.StringVar(&FLAGS.FIELD_SEPARATOR, "field-separator", ",", "Field separator used in command line params")
 	flag.StringVar(&FLAGS.FILTER_SEPARATOR, "filter-separator", ":", "Filter separator used in filters")
 
+	flag.StringVar(&FLAGS.DIAL, "dial", "", "address of remote sybild to query")
+
 	FLAGS.UPDATE_TABLE_INFO = false
 	FLAGS.SKIP_OUTLIERS = true
 	FLAGS.SAMPLES = false

src/lib/query_spec.go

@@ -75,7 +75,6 @@ type Grouping struct {
 
 type Aggregation struct {
 	Op       string
-	op_id    int
 	Name     string
 	name_id  int16
 	HistType string
@@ -219,12 +218,8 @@ func (t *Table) Aggregation(name string, op string) Aggregation {
 	col_id := t.get_key_id(name)
 
 	agg := Aggregation{Name: name, name_id: col_id, Op: op}
-	if op == "avg" {
-		agg.op_id = OP_AVG
-	}
 
 	if op == "hist" {
-		agg.op_id = OP_HIST
 		agg.HistType = "basic"
 		if FLAGS.LOG_HIST {
 			agg.HistType = "multi"
@@ -236,10 +231,6 @@ func (t *Table) Aggregation(name string, op string) Aggregation {
 		}
 	}
 
-	if op == DISTINCT_STR {
-		agg.op_id = OP_DISTINCT
-	}
-
 	_, ok := t.IntInfo[col_id]
 	if !ok {
 		// TODO: tell our table we need to load all records!

src/lib/testdata/TestDecodeGoldenFiles/flag_defs.golden.json.b64

@@ -1 +1 @@
-eyJPUCI6ImF2ZyIsIlBSSU5UIjp0cnVlLCJFWFBPUlQiOmZhbHNlLCJMSVNUX1RBQkxFUyI6ZmFsc2UsIkRFQ09ERV9GTEFHUyI6ZmFsc2UsIkVOQ09ERV9GTEFHUyI6ZmFsc2UsIkVOQ09ERV9SRVNVTFRTIjpmYWxzZSwiSU5UX0ZJTFRFUlMiOiIiLCJTVFJfRklMVEVSUyI6IiIsIlNUUl9SRVBMQUNFIjoiIiwiU0VUX0ZJTFRFUlMiOiIiLCJJTlRTIjoiZm9vLGJhciIsIlNUUlMiOiIiLCJHUk9VUFMiOiJhLGIsYyIsIkRJU1RJTkNUIjoiIiwiQUREX1JFQ09SRFMiOjAsIlRJTUUiOmZhbHNlLCJUSU1FX0NPTCI6InRpbWUiLCJUSU1FX0JVQ0tFVCI6MzYwMCwiSElTVF9CVUNLRVQiOjAsIkhEUl9ISVNUIjpmYWxzZSwiTE9HX0hJU1QiOmZhbHNlLCJGSUVMRF9TRVBBUkFUT1IiOiIsIiwiRklMVEVSX1NFUEFSQVRPUiI6IjoiLCJQUklOVF9LRVlTIjpmYWxzZSwiTE9BRF9BTkRfUVVFUlkiOnRydWUsIkxPQURfVEhFTl9RVUVSWSI6ZmFsc2UsIlJFQURfSU5HRVNUSU9OX0xPRyI6ZmFsc2UsIlJFQURfUk9XU1RPUkUiOmZhbHNlLCJTS0lQX0NPTVBBQ1QiOmZhbHNlLCJQUk9GSUxFIjpmYWxzZSwiUFJPRklMRV9NRU0iOmZhbHNlLCJSRUNZQ0xFX01FTSI6dHJ1ZSwiQ0FDSEVEX1FVRVJJRVMiOmZhbHNlLCJXRUlHSFRfQ09MIjoiIiwiTElNSVQiOjEwMCwiREVCVUciOmZhbHNlLCJKU09OIjpmYWxzZSwiR0MiOnRydWUsIkRJUiI6Ii4vZGIvIiwiU09SVCI6IiRDT1VOVCIsIlBSVU5FX0JZIjoiJENPVU5UIiwiVEFCTEUiOiJ0ZXN0YWJsZSIsIlBSSU5UX0lORk8iOmZhbHNlLCJTQU1QTEVTIjpmYWxzZSwiVVBEQVRFX1RBQkxFX0lORk8iOmZhbHNlLCJTS0lQX09VVExJRVJTIjp0cnVlfQ==
+eyJPUCI6ImF2ZyIsIlBSSU5UIjp0cnVlLCJFWFBPUlQiOmZhbHNlLCJMSVNUX1RBQkxFUyI6ZmFsc2UsIkRFQ09ERV9GTEFHUyI6ZmFsc2UsIkVOQ09ERV9GTEFHUyI6ZmFsc2UsIkVOQ09ERV9SRVNVTFRTIjpmYWxzZSwiSU5UX0ZJTFRFUlMiOiIiLCJTVFJfRklMVEVSUyI6IiIsIlNUUl9SRVBMQUNFIjoiIiwiU0VUX0ZJTFRFUlMiOiIiLCJJTlRTIjoiZm9vLGJhciIsIlNUUlMiOiIiLCJHUk9VUFMiOiJhLGIsYyIsIkRJU1RJTkNUIjoiIiwiQUREX1JFQ09SRFMiOjAsIlRJTUUiOmZhbHNlLCJUSU1FX0NPTCI6InRpbWUiLCJUSU1FX0JVQ0tFVCI6MzYwMCwiSElTVF9CVUNLRVQiOjAsIkhEUl9ISVNUIjpmYWxzZSwiTE9HX0hJU1QiOmZhbHNlLCJGSUVMRF9TRVBBUkFUT1IiOiIsIiwiRklMVEVSX1NFUEFSQVRPUiI6IjoiLCJQUklOVF9LRVlTIjpmYWxzZSwiTE9BRF9BTkRfUVVFUlkiOnRydWUsIkxPQURfVEhFTl9RVUVSWSI6ZmFsc2UsIlJFQURfSU5HRVNUSU9OX0xPRyI6ZmFsc2UsIlJFQURfUk9XU1RPUkUiOmZhbHNlLCJTS0lQX0NPTVBBQ1QiOmZhbHNlLCJQUk9GSUxFIjpmYWxzZSwiUFJPRklMRV9NRU0iOmZhbHNlLCJSRUNZQ0xFX01FTSI6dHJ1ZSwiQ0FDSEVEX1FVRVJJRVMiOmZhbHNlLCJXRUlHSFRfQ09MIjoiIiwiTElNSVQiOjEwMCwiREVCVUciOmZhbHNlLCJKU09OIjpmYWxzZSwiR0MiOnRydWUsIkRJUiI6Ii4vZGIvIiwiU09SVCI6IiRDT1VOVCIsIlBSVU5FX0JZIjoiJENPVU5UIiwiVEFCTEUiOiJ0ZXN0YWJsZSIsIlBSSU5UX0lORk8iOmZhbHNlLCJTQU1QTEVTIjpmYWxzZSwiVVBEQVRFX1RBQkxFX0lORk8iOmZhbHNlLCJTS0lQX09VVExJRVJTIjp0cnVlLCJESUFMIjoiIn0=

src/sybild/doc.go

@@ -0,0 +1,6 @@
+// Package sybild holds the implementation of the sybil grpc service.
+//
+// A sybil server can be started with the `serve` subcommand.
+//
+// Sybil queries are sent to a remote by supplying a value to the `-dial` argument.
+package sybild