You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2021/01/31 00:17:51 UTC
[GitHub] [skywalking-satellite] nic-chen commented on a change in pull request #22: add memory queue
nic-chen commented on a change in pull request #22:
URL: https://github.com/apache/skywalking-satellite/pull/22#discussion_r567342027
##########
File path: plugins/queue/memory/queue.go
##########
@@ -0,0 +1,121 @@
+// 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
+ discardOldest = "DISCARD_OLDEST"
+ discardLatest = "DISCARD_LATEST"
+)
+
+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 DISCARD_OLDEST and DISCARD_LATEST.
+discard_strategy: DISCARD_OLDEST
+`
+}
+
+func (q *Queue) Initialize() error {
+ if q.EventBufferSize <= 0 {
+ return fmt.Errorf("the size of the memory queue must be positive")
+ }
+ if q.DiscardStrategy != discardLatest && q.DiscardStrategy != discardOldest {
+ 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) Enqueue(e *protocol.Event) error {
+ if q.isFull() {
+ switch q.DiscardStrategy {
+ case discardLatest:
+ return api.ErrFull
+ case discardOldest:
+ atomic.AddInt64(&q.readPos, 1)
+ }
+ } else {
+ atomic.AddInt64(&q.count, 1)
+ }
+ q.queue[q.writePos%q.count] = e
+ q.writePos++
+ return nil
+}
+
+func (q *Queue) Dequeue() (*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) isEmpty() bool {
+ return q.count == 0
Review comment:
should we use atomic load here？
##########
File path: plugins/queue/memory/queue.go
##########
@@ -0,0 +1,121 @@
+// 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
+ discardOldest = "DISCARD_OLDEST"
+ discardLatest = "DISCARD_LATEST"
+)
+
+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 DISCARD_OLDEST and DISCARD_LATEST.
+discard_strategy: DISCARD_OLDEST
+`
+}
+
+func (q *Queue) Initialize() error {
+ if q.EventBufferSize <= 0 {
+ return fmt.Errorf("the size of the memory queue must be positive")
+ }
+ if q.DiscardStrategy != discardLatest && q.DiscardStrategy != discardOldest {
+ 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) Enqueue(e *protocol.Event) error {
+ if q.isFull() {
+ switch q.DiscardStrategy {
+ case discardLatest:
+ return api.ErrFull
+ case discardOldest:
+ atomic.AddInt64(&q.readPos, 1)
+ }
+ } else {
+ atomic.AddInt64(&q.count, 1)
+ }
+ q.queue[q.writePos%q.count] = e
+ q.writePos++
+ return nil
+}
+
+func (q *Queue) Dequeue() (*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) isEmpty() bool {
+ return q.count == 0
+}
+
+func (q *Queue) isFull() bool {
+ return q.count == q.EventBufferSize
Review comment:
ditto
##########
File path: plugins/queue/memory/queue.go
##########
@@ -0,0 +1,121 @@
+// 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
+ discardOldest = "DISCARD_OLDEST"
+ discardLatest = "DISCARD_LATEST"
+)
+
+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 DISCARD_OLDEST and DISCARD_LATEST.
+discard_strategy: DISCARD_OLDEST
+`
+}
+
+func (q *Queue) Initialize() error {
+ if q.EventBufferSize <= 0 {
+ return fmt.Errorf("the size of the memory queue must be positive")
Review comment:
we should use `errors.New` for errors with simple static strings
##########
File path: 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_Enqueue_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": discardOldest,
+ },
+ wantErr: false,
+ },
+ {
+ name: "test_lost_the_new_one_discard_strategy",
+ args: map[string]interface{}{
+ "event_buffer_size": num,
+ "discard_strategy": discardLatest,
+ },
+ 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++ {
Review comment:
we should have test case about exceed the buffer size
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org