Skip to content

Commit 53a99d5

Browse files
davit-yisaachier
davit-y
authored and
isaachier
committed
Add Ingester Main, Builder, and Flags (jaegertracing#952)
1 parent 15f1027 commit 53a99d5

File tree

4 files changed

+360
-0
lines changed

4 files changed

+360
-0
lines changed

‎cmd/ingester/app/builder/builder.go

+79
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package builder
16+
17+
import (
18+
"fmt"
19+
20+
"github.com/uber/jaeger-lib/metrics"
21+
"go.uber.org/zap"
22+
23+
"github.com/jaegertracing/jaeger/cmd/ingester/app"
24+
"github.com/jaegertracing/jaeger/cmd/ingester/app/consumer"
25+
"github.com/jaegertracing/jaeger/cmd/ingester/app/processor"
26+
kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
27+
"github.com/jaegertracing/jaeger/plugin/storage/kafka"
28+
"github.com/jaegertracing/jaeger/storage/spanstore"
29+
)
30+
31+
// CreateConsumer creates a new span consumer for the ingester
32+
func CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWriter spanstore.Writer, options app.Options) (*consumer.Consumer, error) {
33+
var unmarshaller kafka.Unmarshaller
34+
if options.Encoding == app.EncodingJSON {
35+
unmarshaller = kafka.NewJSONUnmarshaller()
36+
} else if options.Encoding == app.EncodingProto {
37+
unmarshaller = kafka.NewProtobufUnmarshaller()
38+
} else {
39+
return nil, fmt.Errorf(`encoding '%s' not recognised, use one of ("%s" or "%s")`,
40+
options.Encoding, app.EncodingProto, app.EncodingJSON)
41+
}
42+
43+
spParams := processor.SpanProcessorParams{
44+
Writer: spanWriter,
45+
Unmarshaller: unmarshaller,
46+
}
47+
spanProcessor := processor.NewSpanProcessor(spParams)
48+
49+
consumerConfig := kafkaConsumer.Configuration{
50+
Brokers: options.Brokers,
51+
Topic: options.Topic,
52+
GroupID: options.GroupID,
53+
}
54+
saramaConsumer, err := consumerConfig.NewConsumer()
55+
if err != nil {
56+
return nil, err
57+
}
58+
59+
factoryParams := consumer.ProcessorFactoryParams{
60+
Topic: options.Topic,
61+
Parallelism: options.Parallelism,
62+
SaramaConsumer: saramaConsumer,
63+
BaseProcessor: spanProcessor,
64+
Logger: logger,
65+
Factory: metricsFactory,
66+
}
67+
processorFactory, err := consumer.NewProcessorFactory(factoryParams)
68+
if err != nil {
69+
return nil, err
70+
}
71+
72+
consumerParams := consumer.Params{
73+
InternalConsumer: saramaConsumer,
74+
ProcessorFactory: *processorFactory,
75+
Factory: metricsFactory,
76+
Logger: logger,
77+
}
78+
return consumer.New(consumerParams)
79+
}

‎cmd/ingester/app/flags.go

+97
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package app
16+
17+
import (
18+
"flag"
19+
"fmt"
20+
"strconv"
21+
"strings"
22+
23+
"github.com/spf13/viper"
24+
25+
kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
26+
)
27+
28+
const (
29+
// EncodingJSON indicates spans are encoded as a json byte array
30+
EncodingJSON = "json"
31+
// EncodingProto indicates spans are encoded as a protobuf byte array
32+
EncodingProto = "protobuf"
33+
34+
// ConfigPrefix is a prefix fro the ingester flags
35+
ConfigPrefix = "ingester"
36+
// SuffixBrokers is a suffix for the brokers flag
37+
SuffixBrokers = ".brokers"
38+
// SuffixTopic is a suffix for the topic flag
39+
SuffixTopic = ".topic"
40+
// SuffixGroupID is a suffix for the group-id flag
41+
SuffixGroupID = ".group-id"
42+
// SuffixParallelism is a suffix for the parallelism flag
43+
SuffixParallelism = ".parallelism"
44+
// SuffixEncoding is a suffix for the encoding flag
45+
SuffixEncoding = ".encoding"
46+
47+
// DefaultBroker is the default kafka broker
48+
DefaultBroker = "127.0.0.1:9092"
49+
// DefaultTopic is the default kafka topic
50+
DefaultTopic = "jaeger-spans"
51+
// DefaultGroupID is the default consumer Group ID
52+
DefaultGroupID = "jaeger-ingester"
53+
// DefaultParallelism is the default parallelism for the span processor
54+
DefaultParallelism = 1000
55+
// DefaultEncoding is the default span encoding
56+
DefaultEncoding = EncodingProto
57+
)
58+
59+
// Options stores the configuration options for the Ingester
60+
type Options struct {
61+
kafkaConsumer.Configuration
62+
Parallelism int
63+
Encoding string
64+
}
65+
66+
// AddFlags adds flags for Builder
67+
func AddFlags(flagSet *flag.FlagSet) {
68+
flagSet.String(
69+
ConfigPrefix+SuffixBrokers,
70+
DefaultBroker,
71+
"The comma-separated list of kafka brokers. i.e. '127.0.0.1:9092,0.0.0:1234'")
72+
flagSet.String(
73+
ConfigPrefix+SuffixTopic,
74+
DefaultTopic,
75+
"The name of the kafka topic to consume from")
76+
flagSet.String(
77+
ConfigPrefix+SuffixGroupID,
78+
DefaultGroupID,
79+
"The Consumer Group that ingester will be consuming on behalf of")
80+
flagSet.String(
81+
ConfigPrefix+SuffixParallelism,
82+
strconv.Itoa(DefaultParallelism),
83+
"The number of messages to process in parallel")
84+
flagSet.String(
85+
ConfigPrefix+SuffixEncoding,
86+
DefaultEncoding,
87+
fmt.Sprintf(`The encoding of spans ("%s" or "%s") consumed from kafka`, EncodingProto, EncodingJSON))
88+
}
89+
90+
// InitFromViper initializes Builder with properties from viper
91+
func (o *Options) InitFromViper(v *viper.Viper) {
92+
o.Brokers = strings.Split(v.GetString(ConfigPrefix+SuffixBrokers), ",")
93+
o.Topic = v.GetString(ConfigPrefix + SuffixTopic)
94+
o.GroupID = v.GetString(ConfigPrefix + SuffixGroupID)
95+
o.Parallelism = v.GetInt(ConfigPrefix + SuffixParallelism)
96+
o.Encoding = v.GetString(ConfigPrefix + SuffixEncoding)
97+
}

‎cmd/ingester/app/flags_test.go

+54
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package app
16+
17+
import (
18+
"testing"
19+
20+
"github.com/stretchr/testify/assert"
21+
22+
"github.com/jaegertracing/jaeger/pkg/config"
23+
)
24+
25+
func TestOptionsWithFlags(t *testing.T) {
26+
o := &Options{}
27+
v, command := config.Viperize(AddFlags)
28+
command.ParseFlags([]string{
29+
"--ingester.topic=topic1",
30+
"--ingester.brokers=127.0.0.1:9092,0.0.0:1234",
31+
"--ingester.group-id=group1",
32+
"--ingester.parallelism=5",
33+
"--ingester.encoding=json"})
34+
o.InitFromViper(v)
35+
36+
assert.Equal(t, "topic1", o.Topic)
37+
assert.Equal(t, []string{"127.0.0.1:9092", "0.0.0:1234"}, o.Brokers)
38+
assert.Equal(t, "group1", o.GroupID)
39+
assert.Equal(t, 5, o.Parallelism)
40+
assert.Equal(t, EncodingJSON, o.Encoding)
41+
}
42+
43+
func TestFlagDefaults(t *testing.T) {
44+
o := &Options{}
45+
v, command := config.Viperize(AddFlags)
46+
command.ParseFlags([]string{})
47+
o.InitFromViper(v)
48+
49+
assert.Equal(t, DefaultTopic, o.Topic)
50+
assert.Equal(t, []string{DefaultBroker}, o.Brokers)
51+
assert.Equal(t, DefaultGroupID, o.GroupID)
52+
assert.Equal(t, DefaultParallelism, o.Parallelism)
53+
assert.Equal(t, DefaultEncoding, o.Encoding)
54+
}

‎cmd/ingester/main.go

+130
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package main
16+
17+
import (
18+
"fmt"
19+
"io"
20+
"log"
21+
"os"
22+
"os/signal"
23+
"syscall"
24+
25+
"github.com/spf13/cobra"
26+
"github.com/spf13/viper"
27+
"go.uber.org/zap"
28+
29+
"github.com/jaegertracing/jaeger/cmd/env"
30+
"github.com/jaegertracing/jaeger/cmd/flags"
31+
"github.com/jaegertracing/jaeger/cmd/ingester/app"
32+
"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
33+
"github.com/jaegertracing/jaeger/pkg/config"
34+
pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
35+
"github.com/jaegertracing/jaeger/pkg/version"
36+
"github.com/jaegertracing/jaeger/plugin/storage"
37+
)
38+
39+
func main() {
40+
var signalsChannel = make(chan os.Signal, 0)
41+
signal.Notify(signalsChannel, os.Interrupt, syscall.SIGTERM)
42+
43+
storageFactory, err := storage.NewFactory(storage.FactoryConfigFromEnvAndCLI(os.Args, os.Stderr))
44+
if err != nil {
45+
log.Fatalf("Cannot initialize storage factory: %v", err)
46+
}
47+
48+
v := viper.New()
49+
command := &cobra.Command{
50+
Use: "jaeger-ingester",
51+
Short: "Jaeger ingester consumes from Kafka and writes to storage",
52+
Long: `Jaeger ingester consumes spans from a particular Kafka topic and writes them to all configured storage types.`,
53+
RunE: func(cmd *cobra.Command, args []string) error {
54+
err := flags.TryLoadConfigFile(v)
55+
if err != nil {
56+
return err
57+
}
58+
59+
sFlags := new(flags.SharedFlags).InitFromViper(v)
60+
logger, err := sFlags.NewLogger(zap.NewProductionConfig())
61+
if err != nil {
62+
return err
63+
}
64+
hc, err := sFlags.NewHealthCheck(logger)
65+
if err != nil {
66+
logger.Fatal("Could not start the health check server.", zap.Error(err))
67+
}
68+
69+
mBldr := new(pMetrics.Builder).InitFromViper(v)
70+
baseFactory, err := mBldr.CreateMetricsFactory("jaeger")
71+
if err != nil {
72+
logger.Fatal("Cannot create metrics factory.", zap.Error(err))
73+
}
74+
metricsFactory := baseFactory.Namespace("ingester", nil)
75+
76+
storageFactory.InitFromViper(v)
77+
if err := storageFactory.Initialize(baseFactory, logger); err != nil {
78+
logger.Fatal("Failed to init storage factory", zap.Error(err))
79+
}
80+
spanWriter, err := storageFactory.CreateSpanWriter()
81+
if err != nil {
82+
logger.Fatal("Failed to create span writer", zap.Error(err))
83+
}
84+
85+
options := app.Options{}
86+
options.InitFromViper(v)
87+
consumer, err := builder.CreateConsumer(logger, metricsFactory, spanWriter, options)
88+
if err != nil {
89+
logger.Fatal("Unable to create consumer", zap.Error(err))
90+
}
91+
consumer.Start()
92+
93+
hc.Ready()
94+
select {
95+
case <-signalsChannel:
96+
logger.Info("Jaeger Ingester is starting to close")
97+
err := consumer.Close()
98+
if err != nil {
99+
logger.Error("Failed to close consumer", zap.Error(err))
100+
}
101+
if closer, ok := spanWriter.(io.Closer); ok {
102+
err := closer.Close()
103+
if err != nil {
104+
logger.Error("Failed to close span writer", zap.Error(err))
105+
}
106+
}
107+
logger.Info("Jaeger Ingester has finished closing")
108+
}
109+
return nil
110+
},
111+
}
112+
113+
command.AddCommand(version.Command())
114+
command.AddCommand(env.Command())
115+
116+
config.AddFlags(
117+
v,
118+
command,
119+
flags.AddConfigFileFlag,
120+
flags.AddFlags,
121+
storageFactory.AddFlags,
122+
pMetrics.AddFlags,
123+
app.AddFlags,
124+
)
125+
126+
if err := command.Execute(); err != nil {
127+
fmt.Println(err.Error())
128+
os.Exit(1)
129+
}
130+
}

0 commit comments

Comments
 (0)