<commit_before>// Copyright (C) 2016  <NAME>, Inc.
// Use of this source code is governed by the Apache License 2.0
// that can be found in the COPYING file.

// The occlient tool is a client for the gRPC service for getting and setting the
// OpenConfig configuration and state of a network device.
package main

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"strconv"
	"strings"
	"sync"

	"golang.org/x/net/context"

	"github.com/Shopify/sarama"
	"github.com/aristanetworks/goarista/kafka"
	"github.com/aristanetworks/goarista/kafka/openconfig"
	"github.com/aristanetworks/goarista/kafka/producer"

	"github.com/aristanetworks/glog"
	pb "github.com/aristanetworks/goarista/openconfig"
	"github.com/golang/protobuf/proto"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

const (
	defaultPort   = "6042"
	addrsFlagName = "addrs"
)

var addrsFlag = flag.String(addrsFlagName, "localhost:"+defaultPort,
	"Addresses of the OpenConfig servers (comma-separated)")

var certFileFlag = flag.String("certfile", "",
	"Path to client TLS certificate file")

var keyFileFlag = flag.String("keyfile", "",
	"Path to client TLS private key file")

var caFileFlag = flag.String("cafile", "",
	"Path to server TLS certificate file")

var jsonOutputFlag = flag.Bool("json", false,
	"Print the output in JSON instead of protobuf")

var subscribeFlag = flag.String("subscribe", "",
	"Comma-separated list of paths to subscribe to upon connecting to the server")

var usernameFlag = flag.String("username", "",
	"Username to authenticate with")

var passwordFlag = flag.String("password", "",
	"Password to authenticate with")

var kafkaKeysFlag = flag.String("kafkakey", "",
	"Keys for kafka messages (comma-separated, default: the value of -"+
		addrsFlagName)

type client struct {
	addr          string
	client        pb.OpenConfigClient
	ctx           context.Context
	kafkaProducer producer.Producer
}

func newClient(addr string, opts *[]grpc.DialOption, username, password string,
	kafkaAddresses []string, kafkaTopic, kafkaKey string) (*client, error) {
	c := &client{
		addr: addr,
	}
	conn, err := grpc.Dial(addr, *opts...)
	if err != nil {
		return nil, err
	}
	glog.Infof("Connected to %s", addr)
	c.client = pb.NewOpenConfigClient(conn)
	c.ctx = context.Background()
	if username != "" {
		c.ctx = metadata.NewContext(c.ctx, metadata.Pairs(
			"username", username,
			"password", password))
	}
	if len(kafkaAddresses) == 0 {
		return c, nil
	}
	kafkaConfig := sarama.NewConfig()
	hostname, err := os.Hostname()
	if err != nil {
		hostname = ""
	}
	kafkaConfig.ClientID = hostname
	kafkaConfig.Producer.Compression = sarama.CompressionSnappy
	kafkaConfig.Producer.Return.Successes = true

	kafkaClient, err := sarama.NewClient(kafkaAddresses, kafkaConfig)
	if err != nil {
		glog.Fatalf("Failed to create Kafka client: %s", err)
	}
	kafkaChan := make(chan proto.Message)
	key := sarama.StringEncoder(kafkaKey)
	c.kafkaProducer, err = producer.New(kafkaTopic, kafkaChan, kafkaClient, key,
		openconfig.MessageEncoder)
	if err != nil {
		return nil, fmt.Errorf("Failed to create Kafka producer: %s", err)
	}
	return c, nil
}

func (c *client) run(wg sync.WaitGroup, subscribePaths []string) error {
	defer wg.Done()
	stream, err := c.client.Subscribe(c.ctx)
	if err != nil {
		glog.Fatalf("Subscribe failed: %s", err)
	}
	for _, path := range subscribePaths {
		sub := &pb.SubscribeRequest{
			Request: &pb.SubscribeRequest_Subscribe{
				Subscribe: &pb.SubscriptionList{
					Subscription: []*pb.Subscription{
						&pb.Subscription{
							Path: &pb.Path{Element: strings.Split(path, "/")},
						},
					},
				},
			},
		}

		err = stream.Send(sub)
		if err != nil {
			return err
		}
	}
	if c.kafkaProducer != nil {
		go c.kafkaProducer.Run()
	}
	for {
		resp, err := stream.Recv()
		if err != nil {
			if err != io.EOF {
				glog.Fatalf("Error received from the %s: %s", c.addr, err)
			}
		}
		var respTxt string
		if *jsonOutputFlag {
			respTxt = jsonify(resp)
		} else {
			respTxt = proto.MarshalTextString(resp)
		}
		log := glog.Info
		if resp.GetHeartbeat() != nil {
			log = glog.V(1).Info // Log heartbeats with verbose logging only.
		}
		log(respTxt)
		if c.kafkaProducer != nil {
			c.kafkaProducer.Write(resp)
		}
	}
}

func main() {
	flag.Parse()
	var opts []grpc.DialOption
	if *caFileFlag != "" || *certFileFlag != "" {
		config := &tls.Config{}
		if *caFileFlag != "" {
			b, err := ioutil.ReadFile(*caFileFlag)
			if err != nil {
				glog.Fatal(err)
			}
			cp := x509.NewCertPool()
			if !cp.AppendCertsFromPEM(b) {<commit_msg>cmd/occlient: log responses only if there is no consumer<commit_after>// Copyright (C) 2016  Arista Networks, Inc.
// Use of this source code is governed by the Apache License 2.0
// that can be found in the COPYING file.

// The occlient tool is a client for the gRPC service for getting and setting the
// OpenConfig configuration and state of a network device.
package main

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"strconv"
	"strings"
	"sync"

	"golang.org/x/net/context"

	"github.com/Shopify/sarama"
	"github.com/aristanetworks/goarista/kafka"
	"github.com/aristanetworks/goarista/kafka/openconfig"
	"github.com/aristanetworks/goarista/kafka/producer"

	"github.com/aristanetworks/glog"
	pb "github.com/aristanetworks/goarista/openconfig"
	"github.com/golang/protobuf/proto"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

const (
	defaultPort   = "6042"
	addrsFlagName = "addrs"
)

var addrsFlag = flag.String(addrsFlagName, "localhost:"+defaultPort,
	"Addresses of the OpenConfig servers (comma-separated)")

var certFileFlag = flag.String("certfile", "",
	"Path to client TLS certificate file")

var keyFileFlag = flag.String("keyfile", "",
	"Path to client TLS private key file")

var caFileFlag = flag.String("cafile", "",
	"Path to server TLS certificate file")

var jsonOutputFlag = flag.Bool("json", false,
	"Print the output in JSON instead of protobuf")

var subscribeFlag = flag.String("subscribe", "",
	"Comma-separated list of paths to subscribe to upon connecting to the server")

var usernameFlag = flag.String("username", "",
	"Username to authenticate with")

var passwordFlag = flag.String("password", "",
	"Password to authenticate with")

var kafkaKeysFlag = flag.String("kafkakey", "",
	"Keys for kafka messages (comma-separated, default: the value of -"+
		addrsFlagName)

type client struct {
	addr          string
	client        pb.OpenConfigClient
	ctx           context.Context
	kafkaProducer producer.Producer
}

func newClient(addr string, opts *[]grpc.DialOption, username, password string,
	kafkaAddresses []string, kafkaTopic, kafkaKey string) (*client, error) {
	c := &client{
		addr: addr,
	}
	conn, err := grpc.Dial(addr, *opts...)
	if err != nil {
		return nil, err
	}
	glog.Infof("Connected to %s", addr)
	c.client = pb.NewOpenConfigClient(conn)
	c.ctx = context.Background()
	if username != "" {
		c.ctx = metadata.NewContext(c.ctx, metadata.Pairs(
			"username", username,
			"password", password))
	}
	if len(kafkaAddresses) == 0 {
		return c, nil
	}
	kafkaConfig := sarama.NewConfig()
	hostname, err := os.Hostname()
	if err != nil {
		hostname = ""
	}
	kafkaConfig.ClientID = hostname
	kafkaConfig.Producer.Compression = sarama.CompressionSnappy
	kafkaConfig.Producer.Return.Successes = true

	kafkaClient, err := sarama.NewClient(kafkaAddresses, kafkaConfig)
	if err != nil {
		glog.Fatalf("Failed to create Kafka client: %s", err)
	}
	kafkaChan := make(chan proto.Message)
	key := sarama.StringEncoder(kafkaKey)
	c.kafkaProducer, err = producer.New(kafkaTopic, kafkaChan, kafkaClient, key,
		openconfig.MessageEncoder)
	if err != nil {
		return nil, fmt.Errorf("Failed to create Kafka producer: %s", err)
	}
	return c, nil
}

func (c *client) run(wg sync.WaitGroup, subscribePaths []string) error {
	defer wg.Done()
	stream, err := c.client.Subscribe(c.ctx)
	if err != nil {
		glog.Fatalf("Subscribe failed: %s", err)
	}
	for _, path := range subscribePaths {
		sub := &pb.SubscribeRequest{
			Request: &pb.SubscribeRequest_Subscribe{
				Subscribe: &pb.SubscriptionList{
					Subscription: []*pb.Subscription{
						&pb.Subscription{
							Path: &pb.Path{Element: strings.Split(path, "/")},
						},
					},
				},
			},
		}

		err = stream.Send(sub)
		if err != nil {
			return err
		}
	}
	if c.kafkaProducer != nil {
		go c.kafkaProducer.Run()
	}
	for {
		resp, err := stream.Recv()
		if err != nil {
			if err != io.EOF {
				glog.Fatalf("Error received from the %s: %s", c.addr, err)
			}
		}
		var respTxt string
		if *jsonOutputFlag {
			respTxt = jsonify(resp)
		} else {
			respTxt = proto.MarshalTextString(resp)
		}
		log := glog.Info
		if resp.GetHeartbeat() != nil {
			log = glog.V(1).Info // Log heartbeats with verbose logging only.
		}
		if c.kafkaProducer != nil {
			c.kafkaProducer.Write(resp)
		} else {
			log(respTxt)
		}
	}
}

func main() {
	flag.Parse()
	var opts []grpc.DialOption
	if *caFileFlag != "" || *certFileFlag != "" {
		config := &tls.Config{}
		if *caFileFlag != "" {
			b, err := ioutil.ReadFile(*caFileFlag)
			if err != nil {
				glog.Fatal(err)
			}
			cp := x509.NewCertPool()
			if !cp.AppendCertsFromPEM(b) {
{'fraction_non_alphanumeric': 0.13779107028412732, 'fraction_numerical': 0.0038453321939756462, 'mean_word_length': 3.7697401935812533, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1834542', 'n_tokens_mistral': 3614, 'n_tokens_neox': 3331, 'n_words': 828}