Skip to content

Commit 012147a

Browse files
author
Davit Yeghshatyan
committed
Add builder, flags, and main
Signed-off-by: Davit Yeghshatyan <davo@uber.com>
1 parent c12a9e2 commit 012147a

File tree

4 files changed

+355
-0
lines changed

4 files changed

+355
-0
lines changed

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

+123
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
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+
"strings"
19+
20+
"github.com/spf13/viper"
21+
"github.com/uber/jaeger-lib/metrics"
22+
"go.uber.org/zap"
23+
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+
const (
32+
// EncodingJSON indicates spans are encoded as a json byte array
33+
EncodingJSON = "json"
34+
// EncodingProto indicates spans are encoded as a protobuf byte array
35+
EncodingProto = "protobuf"
36+
37+
// ConfigPrefix is a prefix fro the ingester flags
38+
ConfigPrefix = "ingester"
39+
// SuffixBrokers is a suffix for the brokers flag
40+
SuffixBrokers = ".brokers"
41+
// SuffixTopic is a suffix for the topic flag
42+
SuffixTopic = ".topic"
43+
// SuffixGroupID is a suffix for the group-id flag
44+
SuffixGroupID = ".group-id"
45+
// SuffixParallelism is a suffix for the parallelism flag
46+
SuffixParallelism = ".parallelism"
47+
// SuffixEncoding is a suffix for the encoding flag
48+
SuffixEncoding = ".encoding"
49+
50+
// DefaultBroker is the default kafka broker
51+
DefaultBroker = "127.0.0.1:9092"
52+
// DefaultTopic is the default kafka topic
53+
DefaultTopic = "jaeger-spans"
54+
// DefaultGroupID is the default consumer Group ID
55+
DefaultGroupID = "jaeger-ingester"
56+
// DefaultParallelism is the default parallelism for the span processor
57+
DefaultParallelism = 1000
58+
// DefaultEncoding is the default span encoding
59+
DefaultEncoding = EncodingJSON
60+
)
61+
62+
// Builder stores the configuration options for the Ingester
63+
type Builder struct {
64+
kafkaConsumer.Configuration
65+
Parallelism int
66+
Encoding string
67+
}
68+
69+
// CreateConsumer creates a new span consumer for the ingester
70+
func (b *Builder) CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWriter spanstore.Writer) (*consumer.Consumer, error) {
71+
var unmarshaller kafka.Unmarshaller
72+
if b.Encoding == EncodingJSON {
73+
unmarshaller = kafka.NewJSONUnmarshaller()
74+
} else {
75+
unmarshaller = kafka.NewProtobufUnmarshaller()
76+
}
77+
78+
spParams := processor.SpanProcessorParams{
79+
Writer: spanWriter,
80+
Unmarshaller: unmarshaller,
81+
}
82+
spanProcessor := processor.NewSpanProcessor(spParams)
83+
84+
consumerConfig := kafkaConsumer.Configuration{
85+
Brokers: b.Brokers,
86+
Topic: b.Topic,
87+
GroupID: b.GroupID,
88+
}
89+
saramaConsumer, err := consumerConfig.NewConsumer()
90+
if err != nil {
91+
return nil, err
92+
}
93+
94+
factoryParams := consumer.ProcessorFactoryParams{
95+
Topic: b.Topic,
96+
Parallelism: b.Parallelism,
97+
SaramaConsumer: saramaConsumer,
98+
BaseProcessor: spanProcessor,
99+
Logger: logger,
100+
Factory: metricsFactory,
101+
}
102+
processorFactory, err := consumer.NewProcessorFactory(factoryParams)
103+
if err != nil {
104+
return nil, err
105+
}
106+
107+
consumerParams := consumer.Params{
108+
InternalConsumer: saramaConsumer,
109+
ProcessorFactory: *processorFactory,
110+
Factory: metricsFactory,
111+
Logger: logger,
112+
}
113+
return consumer.New(consumerParams)
114+
}
115+
116+
// InitFromViper initializes Builder with properties from viper
117+
func (b *Builder) InitFromViper(v *viper.Viper) {
118+
b.Brokers = strings.Split(v.GetString(ConfigPrefix+SuffixBrokers), ",")
119+
b.Topic = v.GetString(ConfigPrefix + SuffixTopic)
120+
b.GroupID = v.GetString(ConfigPrefix + SuffixGroupID)
121+
b.Parallelism = v.GetInt(ConfigPrefix + SuffixParallelism)
122+
b.Encoding = v.GetString(ConfigPrefix + SuffixEncoding)
123+
}

‎cmd/ingester/app/flags.go

+47
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
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+
22+
"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
23+
)
24+
25+
// AddFlags adds flags for Builder
26+
func AddFlags(flagSet *flag.FlagSet) {
27+
flagSet.String(
28+
builder.ConfigPrefix+builder.SuffixBrokers,
29+
builder.DefaultBroker,
30+
"The comma-separated list of kafka brokers. i.e. '127.0.0.1:9092,0.0.0:1234'")
31+
flagSet.String(
32+
builder.ConfigPrefix+builder.SuffixTopic,
33+
builder.DefaultTopic,
34+
"The name of the kafka topic to consume from")
35+
flagSet.String(
36+
builder.ConfigPrefix+builder.SuffixGroupID,
37+
builder.DefaultGroupID,
38+
"The Consumer Group that ingester will be consuming on behalf of")
39+
flagSet.String(
40+
builder.ConfigPrefix+builder.SuffixParallelism,
41+
strconv.Itoa(builder.DefaultParallelism),
42+
"The number of messages to process in parallel")
43+
flagSet.String(
44+
builder.ConfigPrefix+builder.SuffixEncoding,
45+
builder.DefaultEncoding,
46+
fmt.Sprintf(`The encoding of spans ("%s" or "%s") consumed from kafka`, builder.EncodingProto, builder.EncodingJSON))
47+
}

‎cmd/ingester/app/flags_test.go

+55
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
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/cmd/ingester/app/builder"
23+
"github.com/jaegertracing/jaeger/pkg/config"
24+
)
25+
26+
func TestOptionsWithFlags(t *testing.T) {
27+
b := &builder.Builder{}
28+
v, command := config.Viperize(AddFlags)
29+
command.ParseFlags([]string{
30+
"--ingester.topic=topic1",
31+
"--ingester.brokers=127.0.0.1:9092,0.0.0:1234",
32+
"--ingester.group-id=group1",
33+
"--ingester.parallelism=5",
34+
"--ingester.encoding=json"})
35+
b.InitFromViper(v)
36+
37+
assert.Equal(t, "topic1", b.Topic)
38+
assert.Equal(t, []string{"127.0.0.1:9092", "0.0.0:1234"}, b.Brokers)
39+
assert.Equal(t, "group1", b.GroupID)
40+
assert.Equal(t, 5, b.Parallelism)
41+
assert.Equal(t, builder.EncodingJSON, b.Encoding)
42+
}
43+
44+
func TestFlagDefaults(t *testing.T) {
45+
b := &builder.Builder{}
46+
v, command := config.Viperize(AddFlags)
47+
command.ParseFlags([]string{})
48+
b.InitFromViper(v)
49+
50+
assert.Equal(t, builder.DefaultTopic, b.Topic)
51+
assert.Equal(t, []string{builder.DefaultBroker}, b.Brokers)
52+
assert.Equal(t, builder.DefaultGroupID, b.GroupID)
53+
assert.Equal(t, builder.DefaultParallelism, b.Parallelism)
54+
assert.Equal(t, builder.DefaultEncoding, b.Encoding)
55+
}

‎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+
b := builder.Builder{}
86+
b.InitFromViper(v)
87+
consumer, err := b.CreateConsumer(logger, metricsFactory, spanWriter)
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)