cmd

var (
	// ErrNoScrapeConfig is when the provided config omits or has a zero list of scrape configs
	ErrNoScrapeConfig = errors.New("no scrape config found")
	// ErrMultipleScrapeConfig is when the provided config has more than one scrape configs
	ErrMultipleScrapeConfig = errors.New("multiple scrape configs found")
	// ErrNoArgs is when the command line expected an arg but none was provided
	ErrNoArgs = errors.New("no args")
	// ErrMultipleArgs is when the command line expected one arge but multiple were provided
	ErrMultipleArgs = errors.New("multiple arguments in single argument command")
)

var Ingester = &cobra.Command{
	Use:   "ingester",
	Short: "runs the ingester service to consume metrics from kafka into cassandra",
	RunE: func(cmd *cobra.Command, args []string) error {
		log.SetLevel(log.DebugLevel)

		cmd.Flags().VisitAll(func(f *pflag.Flag) {
			viper.BindPFlag(f.Name, f)
		})

		err := cassandra.SetupTables(strings.Split(viper.GetString("cassandra-addrs"), ","), viper.GetString("cassandra-keyspace"))
		if err != nil {
			return err
		}

		sw, err := cassandra.NewSampleWriter(&cassandra.SampleWriterConfig{
			CassandraAddrs: strings.Split(viper.GetString("cassandra-addrs"), ","),
			Keyspace:       viper.GetString("cassandra-keyspace"),
			Timeout:        30 * time.Second,
		})
		if err != nil {
			return err
		}

		source, err := kafka.NewAckSource(&kafka.AckSourceConfig{
			Addrs:     strings.Split(viper.GetString("kafka-addrs"), ","),
			ClientID:  viper.GetString("kafka-client-id"),
			Converter: kafka.DefaultConverter{},
			Topic:     viper.GetString("kafka-topic"),
		})
		if err != nil {
			return err
		}
		prometheus.MustRegister(source)

		i := ingester.NewIngester(&ingester.Config{
			SampleWriter: sw,
			AckSource:    source,
		})
		prometheus.MustRegister(i)
		go func() {
			http.Handle("/metrics", prometheus.Handler())
			http.ListenAndServe(":8080", nil)
		}()
		return i.Run()
	},
}