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 2021/01/30 14:00:45 UTC
[skywalking-satellite] 01/01: add memory queue
This is an automated email from the ASF dual-hosted git repository.
liujiapeng pushed a commit to branch memory-queue
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git
commit 2af8c80eaa8eef1c91ef27958baea9a359167392
Author: Evan <ev...@outlook.com>
AuthorDate: Sat Jan 30 22:00:12 2021 +0800
add memory queue
---
docs/plugin-description.md | 12 ++++
plugins/queue/memory/queue.go | 138 +++++++++++++++++++++++++++++++++++++
plugins/queue/memory/queue_test.go | 120 ++++++++++++++++++++++++++++++++
plugins/queue/queue_repository.go | 4 +-
4 files changed, 273 insertions(+), 1 deletion(-)
diff --git a/docs/plugin-description.md b/docs/plugin-description.md
index 0b7506a..290aea3 100755
--- a/docs/plugin-description.md
+++ b/docs/plugin-description.md
@@ -77,6 +77,18 @@ this is a synchronization Kafka log forwarder.
topic: "log-topic"
```
# api.Queue
+## memory-queue
+### description
+this is a memory queue to buffer the input event.
+### defaultConfig
+```yaml
+# The maximum buffer event size.
+event_buffer_size: 5000
+# The discard strategy when facing the full condition.
+# There are 2 strategies, which are LOST_THE_OLDEST_ONE and LOST_THE_NEW_ONE.
+discard_strategy: LOST_THE_OLDEST_ONE
+```
+# api.Queue
## mmap-queue
### description
this is a memory mapped queue to provide the persistent storage.
diff --git a/plugins/queue/memory/queue.go b/plugins/queue/memory/queue.go
new file mode 100644
index 0000000..a911272
--- /dev/null
+++ b/plugins/queue/memory/queue.go
@@ -0,0 +1,138 @@
+// 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 memory
+
+import (
+ "fmt"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/config"
+ "github.com/apache/skywalking-satellite/internal/satellite/event"
+ "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/protocol/gen-codes/satellite/protocol"
+)
+
+const (
+ Name = "memory-queue"
+ // discard strategy
+ lostOldestOne = "LOST_THE_OLDEST_ONE"
+ lostNewOne = "LOST_THE_NEW_ONE"
+)
+
+type Queue struct {
+ config.CommonFields
+ // config
+ EventBufferSize int64 `mapstructure:"event_buffer_size"` // The maximum buffer event size.
+ DiscardStrategy string `mapstructure:"discard_strategy"` // The discard strategy.
+
+ // components
+ queue []*protocol.Event
+ // The position continuously increasing, but don't worry, it can run for another 1067519911 days at 10W OPS.
+ readPos int64
+ writePos int64
+ count int64
+}
+
+func (q *Queue) Name() string {
+ return Name
+}
+
+func (q *Queue) Description() string {
+ return "this is a memory queue to buffer the input event."
+}
+
+func (q *Queue) DefaultConfig() string {
+ return `
+# The maximum buffer event size.
+event_buffer_size: 5000
+# The discard strategy when facing the full condition.
+# There are 2 strategies, which are LOST_THE_OLDEST_ONE and LOST_THE_NEW_ONE.
+discard_strategy: LOST_THE_OLDEST_ONE
+`
+}
+
+func (q *Queue) Initialize() error {
+ if q.EventBufferSize <= 0 {
+ return fmt.Errorf("the size of the memory queue must be positive")
+ }
+ if q.DiscardStrategy != lostOldestOne && q.DiscardStrategy != lostNewOne {
+ return fmt.Errorf("%s discard strategy is not supported in the memory queue", q.DiscardStrategy)
+ }
+ q.queue = make([]*protocol.Event, q.EventBufferSize)
+ return nil
+}
+
+func (q *Queue) Push(event *protocol.Event) error {
+ if q.isFull() {
+ switch q.DiscardStrategy {
+ case lostNewOne:
+ return api.ErrFull
+ case lostOldestOne:
+ atomic.AddInt64(&q.readPos, 1)
+ }
+ } else {
+ atomic.AddInt64(&q.count, 1)
+ }
+ q.queue[q.writePos%q.count] = event
+ q.writePos++
+ return nil
+}
+
+func (q *Queue) Pop() (*api.SequenceEvent, error) {
+ if q.isEmpty() {
+ return nil, api.ErrEmpty
+ }
+ e := &api.SequenceEvent{
+ Event: q.queue[q.readPos],
+ }
+ atomic.AddInt64(&q.readPos, 1)
+ atomic.AddInt64(&q.count, -1)
+ return e, nil
+
+}
+
+func (q *Queue) Close() error {
+ return nil
+}
+
+func (q *Queue) Ack(_ event.Offset) {
+}
+
+func (q *Queue) addPos(pos int64) int64 {
+ if pos == q.EventBufferSize-1 {
+ return 0
+ }
+ return pos + 1
+}
+
+func (q *Queue) isEmpty() bool {
+ return q.count == 0
+
+}
+
+func (q *Queue) isFull() bool {
+ return q.count == q.EventBufferSize
+}
+
+func (q *Queue) cas(address *int64, addVal int64) {
+ for {
+ if atomic.CompareAndSwapInt64(address, *address, *address+addVal) {
+ return
+ }
+ }
+}
diff --git a/plugins/queue/memory/queue_test.go b/plugins/queue/memory/queue_test.go
new file mode 100644
index 0000000..76ec988
--- /dev/null
+++ b/plugins/queue/memory/queue_test.go
@@ -0,0 +1,120 @@
+// 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 memory
+
+import (
+ "fmt"
+ "reflect"
+ "testing"
+
+ "github.com/apache/skywalking-satellite/internal/pkg/plugin"
+ _ "github.com/apache/skywalking-satellite/internal/satellite/test"
+ "github.com/apache/skywalking-satellite/plugins/queue/api"
+ "github.com/apache/skywalking-satellite/protocol/gen-codes/satellite/protocol"
+)
+
+func initMemoryQueue(cfg plugin.Config) (*Queue, error) {
+ plugin.RegisterPluginCategory(reflect.TypeOf((*api.Queue)(nil)).Elem())
+ plugin.RegisterPlugin(&Queue{})
+ var config plugin.Config = map[string]interface{}{
+ plugin.NameField: Name,
+ }
+ for k, v := range cfg {
+ config[k] = v
+ }
+ q := api.GetQueue(config)
+ if q == nil {
+ return nil, fmt.Errorf("cannot get a memoory queue from the registry")
+ }
+ if err := q.Initialize(); err != nil {
+ return nil, fmt.Errorf("queue cannot initialize: %v", err)
+ }
+ return q.(*Queue), nil
+}
+
+func TestQueue_Push_Strategy(t *testing.T) {
+ const num = 5
+ tests := []struct {
+ name string
+ args plugin.Config
+ wantErr bool
+ }{
+ {
+ name: "test_lost_the_oldest_one_discard_strategy",
+ args: map[string]interface{}{
+ "event_buffer_size": num,
+ "discard_strategy": "LOST_THE_OLDEST_ONE",
+ },
+ wantErr: false,
+ },
+ {
+ name: "test_lost_the_new_one_discard_strategy",
+ args: map[string]interface{}{
+ "event_buffer_size": num,
+ "discard_strategy": "LOST_THE_NEW_ONE",
+ },
+ wantErr: true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ q, err := initMemoryQueue(tt.args)
+ if err != nil {
+ t.Fatalf("cannot init the memory queue: %v", err)
+ }
+ for i := 0; i < num; i++ {
+ if err := q.Push(new(protocol.Event)); err != nil {
+ t.Fatalf("cannot push event to the queue: %v", err)
+ }
+ }
+ if err := q.Push(new(protocol.Event)); (err != nil) != tt.wantErr {
+ t.Errorf("Push() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ })
+ }
+}
+
+func TestQueue_Push(t *testing.T) {
+ const num = 10
+ q, err := initMemoryQueue(map[string]interface{}{
+ "event_buffer_size": num,
+ "discard_strategy": "LOST_THE_NEW_ONE",
+ })
+ if err != nil {
+ t.Fatalf("cannot init the memory queue: %v", err)
+ }
+
+ for i := 0; i < num; i++ {
+ if err := q.Push(new(protocol.Event)); err != nil {
+ t.Fatalf("the push want seuccess but failure: %v", err)
+ }
+ }
+ if err := q.Push(new(protocol.Event)); err == nil {
+ t.Fatalf("the push want failure but success")
+ }
+ for i := 0; i < num; i++ {
+ if e, err := q.Pop(); err != nil {
+ t.Fatalf("the pop want seuccess but failure: %v", err)
+ } else if e == nil {
+ t.Fatalf("the pop want a event but got nil")
+ }
+ }
+ if _, err := q.Pop(); err == nil {
+ t.Fatalf("the pop want error but success: %v", err)
+ }
+}
diff --git a/plugins/queue/queue_repository.go b/plugins/queue/queue_repository.go
index 1c49822..9ddc561 100644
--- a/plugins/queue/queue_repository.go
+++ b/plugins/queue/queue_repository.go
@@ -18,6 +18,7 @@
package queue
import (
+ "github.com/apache/skywalking-satellite/plugins/queue/memory"
"reflect"
"github.com/apache/skywalking-satellite/internal/pkg/plugin"
@@ -29,8 +30,9 @@ import (
func RegisterQueuePlugins() {
plugin.RegisterPluginCategory(reflect.TypeOf((*api.Queue)(nil)).Elem())
queues := []api.Queue{
- &mmap.Queue{},
// Please register the queue plugins at here.
+ new(memory.Queue),
+ new(mmap.Queue),
}
for _, q := range queues {
plugin.RegisterPlugin(q)