You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2020/11/27 12:06:19 UTC
[skywalking-satellite] branch plugin-framework created (now b57c0df)
This is an automated email from the ASF dual-hosted git repository.
liujiapeng pushed a change to branch plugin-framework
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git.
at b57c0df add api && plugin framework registry
This branch includes the following new commits:
new b57c0df add api && plugin framework registry
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking-satellite] 01/01: add api && plugin framework registry
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
liujiapeng pushed a commit to branch plugin-framework
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git
commit b57c0df5e1ca32e349e0ac5bbde0a8f38dac7444
Author: Evan <ev...@outlook.com>
AuthorDate: Fri Nov 27 20:05:48 2020 +0800
add api && plugin framework registry
---
go.mod | 4 +-
go.sum | 5 +-
internal/pkg/api/event.go | 86 ++++++++++++++
internal/pkg/api/forwarder.go | 26 +++++
internal/pkg/api/gatherer.go | 26 +++++
internal/pkg/api/parser.go | 28 +++++
internal/pkg/api/plugin.go | 31 ++++++
internal/pkg/api/processor.go | 26 +++++
internal/pkg/api/queue.go | 45 ++++++++
internal/pkg/logger/log.go | 3 +
internal/satellite/event/struectured_event.go | 68 ++++++++++++
internal/satellite/event/unstruectured_event.go | 68 ++++++++++++
internal/satellite/registry/registry.go | 142 ++++++++++++++++++++++++
13 files changed, 551 insertions(+), 7 deletions(-)
diff --git a/go.mod b/go.mod
index fa4fc28..71fd69d 100644
--- a/go.mod
+++ b/go.mod
@@ -2,6 +2,4 @@ module github.com/apache/skywalking-satellite
go 1.14
-require (
- github.com/sirupsen/logrus v1.7.0
-)
+require github.com/sirupsen/logrus v1.7.0
diff --git a/go.sum b/go.sum
index 4d54e45..4d74a1e 100644
--- a/go.sum
+++ b/go.sum
@@ -1,13 +1,10 @@
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/sirupsen/logrus v1.7.0 h1:ShrD1U9pZB12TX0cVy0DtePoCH97K8EtX+mg7ZARUtM=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
-github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037 h1:YyJpGZS1sBuBCzLAR1VEpK193GlqGZbnPFnPV/5Rsb4=
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/internal/pkg/api/event.go b/internal/pkg/api/event.go
new file mode 100644
index 0000000..7b87c57
--- /dev/null
+++ b/internal/pkg/api/event.go
@@ -0,0 +1,86 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+import (
+ "fmt"
+ "time"
+)
+
+// Event that implement this interface would be allowed to transmit in the Satellite.
+type Event interface {
+ // Name is a identify to distinguish different events.
+ Name() string
+
+ // Meta is a pair of key and value to record meta data, such as labels.
+ Meta() map[string]string
+
+ // Data returns the wrappered data.
+ Data() interface{}
+
+ // Time returns the event time.
+ Time() time.Time
+
+ // IsOutput indicates that the event can exist in the output context when the return value is true.
+ IsOutput() bool
+}
+
+// InputEvent is used in Gatherer to bridge Queue.
+type InputEvent interface {
+ Event
+
+ // ToBytes serialize the event to a byte array.
+ ToBytes() []byte
+
+ // FromBytes deserialize the byte array to an event.
+ FromBytes(bytes []byte) InputEvent
+}
+
+// BatchOutputEvents is batch events to output.
+type BatchOutputEvents struct {
+ // BatchEvents grouped by event name.
+ BatchEvents map[string][]Event
+
+ // The start offset of the batch.
+ StartOffset int64
+
+ // The end offset of the batch.
+ EndOffset int64
+}
+
+// OutputEventContext is a container to store the output context.
+type OutputEventContext struct {
+ context map[string]Event
+}
+
+// Put puts the incoming event into the context when the event allows to output.
+func (c *OutputEventContext) Put(event Event) {
+ if event.IsOutput() {
+ c.context[event.Name()] = event
+ }
+}
+
+// Get returns a event in the context. When the eventName does not exist, a error would be returned.
+func (c *OutputEventContext) Get(eventName string) (Event, error) {
+ e, ok := c.context[eventName]
+ if !ok {
+ err := fmt.Errorf("cannot find the event name in OutputEventContext : %s", eventName)
+ return nil, err
+ }
+ return e, nil
+}
diff --git a/internal/pkg/api/forwarder.go b/internal/pkg/api/forwarder.go
new file mode 100644
index 0000000..7966041
--- /dev/null
+++ b/internal/pkg/api/forwarder.go
@@ -0,0 +1,26 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+// Forwarder is a plugin interface, that defines new forwarders.
+type Forwarder interface {
+ ComponentPlugin
+
+ // Output the batch events to the external output services, such as Kafka MQ and SkyWalking OAP cluster.
+ Output(batch BatchOutputEvents)
+}
diff --git a/internal/pkg/api/gatherer.go b/internal/pkg/api/gatherer.go
new file mode 100644
index 0000000..9b2c0c0
--- /dev/null
+++ b/internal/pkg/api/gatherer.go
@@ -0,0 +1,26 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+// Gatherer is a plugin interface, that defines new gatherers.
+type Gatherer interface {
+ ComponentPlugin
+
+ // Gather gathers the data from the input.
+ Gather() (*InputEvent, error)
+}
diff --git a/internal/pkg/api/parser.go b/internal/pkg/api/parser.go
new file mode 100644
index 0000000..2ed7953
--- /dev/null
+++ b/internal/pkg/api/parser.go
@@ -0,0 +1,28 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+// Parser is a plugin interface, that defines new Parsers for Gatherer plugin.
+type Parser interface {
+
+ // ParseBytes parse the byte buffer into events.
+ ParseBytes(bytes []byte) ([]InputEvent, error)
+
+ // ParseStr parse the string into events.
+ ParseStr(str string) ([]InputEvent, error)
+}
diff --git a/internal/pkg/api/plugin.go b/internal/pkg/api/plugin.go
new file mode 100644
index 0000000..534352d
--- /dev/null
+++ b/internal/pkg/api/plugin.go
@@ -0,0 +1,31 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+import "io"
+
+// ComponentPlugin is an interface to initialize the specific plugin.
+type ComponentPlugin interface {
+ io.Closer
+
+ // Init initialize the specific plugin and would return error when the configuration is error.
+ Init() error
+
+ // Run triggers the specific plugin to work.
+ Run()
+}
diff --git a/internal/pkg/api/processor.go b/internal/pkg/api/processor.go
new file mode 100644
index 0000000..8067023
--- /dev/null
+++ b/internal/pkg/api/processor.go
@@ -0,0 +1,26 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+// Processor is a plugin interface, that defines new pipeline processors.
+type Processor interface {
+ ComponentPlugin
+
+ // Process produces a new event by processing incoming event.
+ Process(in Event) Event
+}
diff --git a/internal/pkg/api/queue.go b/internal/pkg/api/queue.go
new file mode 100644
index 0000000..e3f9067
--- /dev/null
+++ b/internal/pkg/api/queue.go
@@ -0,0 +1,45 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api
+
+// Queue is a plugin interface, that defines new queues.
+type Queue interface {
+ ComponentPlugin
+
+ // Publisher get the only publisher for the current queue.
+ Publisher() QueuePublisher
+
+ // Consumer get the only consumer for the current queue.
+ Consumer() QueueConsumer
+}
+
+// QueuePublisher is a plugin interface, that defines new queue publishers.
+type QueuePublisher interface {
+ ComponentPlugin
+
+ // Enqueue push a inputEvent into the queue.
+ Enqueue(event *InputEvent) error
+}
+
+// QueueConsumer is a plugin interface, that defines new queue consumers.
+type QueueConsumer interface {
+ ComponentPlugin
+
+ // Dequeue pop an event form the Queue. When the queue is empty, the method would be blocked.
+ Dequeue() (*InputEvent, error)
+}
diff --git a/internal/pkg/logger/log.go b/internal/pkg/logger/log.go
index 6021bf6..cc3ae7d 100644
--- a/internal/pkg/logger/log.go
+++ b/internal/pkg/logger/log.go
@@ -62,6 +62,9 @@ func Init(opts ...Option) {
if f.timePattern == "" {
f.timePattern = defaultTimePattern
}
+ if !strings.Contains(f.logPattern, "\n") {
+ f.logPattern = f.logPattern + "\n"
+ }
Log.SetFormatter(f)
})
}
diff --git a/internal/satellite/event/struectured_event.go b/internal/satellite/event/struectured_event.go
new file mode 100644
index 0000000..9d83c63
--- /dev/null
+++ b/internal/satellite/event/struectured_event.go
@@ -0,0 +1,68 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package event
+
+import (
+ "github.com/apache/skywalking-satellite/internal/pkg/api"
+ "time"
+)
+
+type StructuredInputEventToBytesFunc func(event *StructuredInputEvent) []byte
+type BytesToStructuredInputEventFunc func(bytes []byte) *StructuredInputEvent
+
+type StructuredEvent struct {
+ name string
+ timestamp time.Time
+ meta map[string]string
+ data interface{}
+ output bool
+}
+
+type StructuredInputEvent struct {
+ StructuredEvent
+ etb StructuredInputEventToBytesFunc
+ bte BytesToStructuredInputEventFunc
+}
+
+func (s *StructuredEvent) Name() string {
+ return s.name
+}
+
+func (s *StructuredEvent) Meta() map[string]string {
+ return s.meta
+}
+
+func (s *StructuredEvent) Data() interface{} {
+ return &s.data
+}
+
+func (s *StructuredEvent) Time() time.Time {
+ return s.timestamp
+}
+
+func (s *StructuredEvent) IsOutput() bool {
+ return s.output
+}
+
+func (s *StructuredInputEvent) ToBytes() []byte {
+ return s.etb(s)
+}
+
+func (s *StructuredInputEvent) FromBytes(bytes []byte) api.InputEvent {
+ return s.bte(bytes)
+}
diff --git a/internal/satellite/event/unstruectured_event.go b/internal/satellite/event/unstruectured_event.go
new file mode 100644
index 0000000..37ddf6f
--- /dev/null
+++ b/internal/satellite/event/unstruectured_event.go
@@ -0,0 +1,68 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package event
+
+import (
+ "github.com/apache/skywalking-satellite/internal/pkg/api"
+ "time"
+)
+
+type UnstructuredInputEventToBytesFunc func(event *UnstructuredInputEvent) []byte
+type BytesToUnstructuredInputEventFunc func(bytes []byte) *UnstructuredInputEvent
+
+type UnstructuredEvent struct {
+ name string
+ timestamp time.Time
+ meta map[string]string
+ data map[string]interface{}
+ output bool
+}
+
+type UnstructuredInputEvent struct {
+ UnstructuredEvent
+ etb UnstructuredInputEventToBytesFunc
+ bte BytesToUnstructuredInputEventFunc
+}
+
+func (s *UnstructuredInputEvent) Name() string {
+ return s.name
+}
+
+func (s *UnstructuredInputEvent) Meta() map[string]string {
+ return s.meta
+}
+
+func (s *UnstructuredInputEvent) Data() interface{} {
+ return &s.data
+}
+
+func (s *UnstructuredInputEvent) Time() time.Time {
+ return s.timestamp
+}
+
+func (s *UnstructuredInputEvent) IsOutput() bool {
+ return s.output
+}
+
+func (s *UnstructuredInputEvent) ToBytes() []byte {
+ return s.etb(s)
+}
+
+func (s *UnstructuredInputEvent) FromBytes(bytes []byte) api.InputEvent {
+ return s.bte(bytes)
+}
diff --git a/internal/satellite/registry/registry.go b/internal/satellite/registry/registry.go
new file mode 100644
index 0000000..87199d6
--- /dev/null
+++ b/internal/satellite/registry/registry.go
@@ -0,0 +1,142 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package registry
+
+import (
+ "fmt"
+ "github.com/apache/skywalking-satellite/internal/pkg/api"
+ "github.com/apache/skywalking-satellite/internal/pkg/logger"
+)
+
+// The creator reg.
+type registry struct {
+ gathererCreatorRegistry map[string]GathererCreator
+ queueCreatorRegistry map[string]QueueCreator
+ processorCreatorRegistry map[string]ProcessorCreator
+ forwarderCreatorRegistry map[string]ForwarderCreator
+ parserCreatorRegistry map[string]ParserCreator
+}
+
+// GathererCreator creates a Gather according to the config.
+type GathererCreator func(config map[string]interface{}) (api.Gatherer, error)
+
+// QueueCreator creates a Queue according to the config.
+type QueueCreator func(config map[string]interface{}) (api.Queue, error)
+
+// ProcessorCreator creates a Processor according to the config.
+type ProcessorCreator func(config map[string]interface{}) (api.Processor, error)
+
+// ForwarderCreator creates a forwarder according to the config.
+type ForwarderCreator func(config map[string]interface{}) (api.Forwarder, error)
+
+// ParserCreator creates a parser according to the config.
+type ParserCreator func(config map[string]interface{}) (api.Parser, error)
+
+var reg *registry
+
+// RegisterGatherer registers the gatherType as GathererCreator.
+func RegisterGatherer(gathererType string, creator GathererCreator) {
+ logger.Log.Info(gathererType)
+ reg.gathererCreatorRegistry[gathererType] = creator
+}
+
+// RegisterQueue registers the queueType as QueueCreator.
+func RegisterQueue(queueType string, creator QueueCreator) {
+ reg.queueCreatorRegistry[queueType] = creator
+}
+
+// RegisterProcessor registers the processorType as ProcessorCreator.
+func RegisterProcessor(processorType string, creator ProcessorCreator) {
+ reg.processorCreatorRegistry[processorType] = creator
+}
+
+// RegisterForwarder registers the forwarderType as forwarderCreator.
+func RegisterForwarder(forwarderType string, creator ForwarderCreator) {
+ reg.forwarderCreatorRegistry[forwarderType] = creator
+}
+
+// CreateGatherer creates a Gatherer according to the gathererType.
+func CreateGatherer(gathererType string, config map[string]interface{}) (api.Gatherer, error) {
+ if c, ok := reg.gathererCreatorRegistry[gathererType]; ok {
+ gatherer, err := c(config)
+ if err != nil {
+ return nil, fmt.Errorf("create gatherer failed: %v", err)
+ }
+ return gatherer, nil
+ }
+ return nil, fmt.Errorf("unsupported gatherer type: %v", gathererType)
+}
+
+// CreateQueue creates a Queue according to the queueType.
+func CreateQueue(queueType string, config map[string]interface{}) (api.Queue, error) {
+ if c, ok := reg.queueCreatorRegistry[queueType]; ok {
+ queue, err := c(config)
+ if err != nil {
+ return nil, fmt.Errorf("create queue failed: %v", err)
+ }
+ return queue, nil
+ }
+ return nil, fmt.Errorf("unsupported queue type: %v", queueType)
+}
+
+// CreateProcessor creates a Processor according to the processorType.
+func CreateProcessor(processorType string, config map[string]interface{}) (api.Processor, error) {
+ if c, ok := reg.processorCreatorRegistry[processorType]; ok {
+ processor, err := c(config)
+ if err != nil {
+ return nil, fmt.Errorf("create processor failed: %v", err)
+ }
+ return processor, nil
+ }
+ return nil, fmt.Errorf("unsupported processor type: %v", processorType)
+}
+
+// CreateForwarder creates a forwarder according to the forwarderType.
+func CreateForwarder(forwarderType string, config map[string]interface{}) (api.Forwarder, error) {
+ if c, ok := reg.forwarderCreatorRegistry[forwarderType]; ok {
+ forwarder, err := c(config)
+ if err != nil {
+ return nil, fmt.Errorf("create forwarder failed: %v", err)
+ }
+ return forwarder, nil
+ }
+ return nil, fmt.Errorf("unsupported forwarder type: %v", forwarderType)
+}
+
+// CreateParser creates a parser according to the parserType.
+func CreateParser(parserType string, config map[string]interface{}) (api.Parser, error) {
+ if c, ok := reg.parserCreatorRegistry[parserType]; ok {
+ parser, err := c(config)
+ if err != nil {
+ return nil, fmt.Errorf("create parser failed: %v", err)
+ }
+ return parser, nil
+ }
+ return nil, fmt.Errorf("unsupported parser type: %v", parserType)
+}
+
+func init() {
+ if reg == nil {
+ reg = ®istry{}
+ reg.gathererCreatorRegistry = make(map[string]GathererCreator)
+ reg.queueCreatorRegistry = make(map[string]QueueCreator)
+ reg.processorCreatorRegistry = make(map[string]ProcessorCreator)
+ reg.forwarderCreatorRegistry = make(map[string]ForwarderCreator)
+ reg.parserCreatorRegistry = make(map[string]ParserCreator)
+ }
+}