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 2022/08/27 15:58:22 UTC
[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #164: Add streaming API and topN aggregator
hanahmily commented on code in PR #164:
URL: https://github.com/apache/skywalking-banyandb/pull/164#discussion_r956591982
##########
pkg/flow/streaming/flow_sliding_window.go:
##########
@@ -0,0 +1,334 @@
+// 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 streaming
+
+import (
+ "container/heap"
+ "context"
+ "math"
+ "sync"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+)
+
+type TriggerResult bool
+
+const (
+ FIRE TriggerResult = true
+ CONTINUE = false
+)
+
+var (
+ _ flow.Operator = (*TumblingTimeWindows)(nil)
+ _ flow.WindowAssigner = (*TumblingTimeWindows)(nil)
+ _ flow.Window = (*timeWindow)(nil)
+
+ _ TriggerContext = (*triggerContext)(nil)
+)
+
+func (f *streamingFlow) Window(w flow.WindowAssigner) flow.WindowedFlow {
+ switch v := w.(type) {
+ case *TumblingTimeWindows:
+ v.errorHandler = f.drainErr
+ f.ops = append(f.ops, v)
+ default:
+ f.drainErr(errors.New("window type is not supported"))
+ }
+
+ return &windowedFlow{
+ f: f,
+ wa: w,
+ }
+}
+
+func (s *windowedFlow) AllowedLateness(lateness time.Duration) flow.Flow {
+ switch v := s.wa.(type) {
+ case *TumblingTimeWindows:
+ v.lateness = lateness.Milliseconds()
+ default:
+ s.f.drainErr(errors.New("lateness is not supported"))
+ }
+ return s.f
+}
+
+type TumblingTimeWindows struct {
+ // internal state of the sliding time window
+ flow.ComponentState
+ // errorHandler is the error handler and set by the streamingFlow
+ errorHandler func(error)
+ // For TumblingTimeWindows
+ // Unit: Millisecond
+ size int64
+ // lateness is the maximum allowed lateness for incoming elements
+ // Unit: Millisecond
+ lateness int64
+
+ // guard snapshots
+ snapshotsMu sync.Mutex
+ snapshots map[int64]flow.AggregationOp
Review Comment:
This map will cause OOM due to the lack of releasing
##########
pkg/flow/streaming/flow_sliding_window.go:
##########
@@ -0,0 +1,334 @@
+// 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 streaming
+
+import (
+ "container/heap"
+ "context"
+ "math"
+ "sync"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+)
+
+type TriggerResult bool
+
+const (
+ FIRE TriggerResult = true
+ CONTINUE = false
+)
+
+var (
+ _ flow.Operator = (*TumblingTimeWindows)(nil)
+ _ flow.WindowAssigner = (*TumblingTimeWindows)(nil)
+ _ flow.Window = (*timeWindow)(nil)
+
+ _ TriggerContext = (*triggerContext)(nil)
+)
+
+func (f *streamingFlow) Window(w flow.WindowAssigner) flow.WindowedFlow {
+ switch v := w.(type) {
+ case *TumblingTimeWindows:
+ v.errorHandler = f.drainErr
+ f.ops = append(f.ops, v)
+ default:
+ f.drainErr(errors.New("window type is not supported"))
+ }
+
+ return &windowedFlow{
+ f: f,
+ wa: w,
+ }
+}
+
+func (s *windowedFlow) AllowedLateness(lateness time.Duration) flow.Flow {
+ switch v := s.wa.(type) {
+ case *TumblingTimeWindows:
+ v.lateness = lateness.Milliseconds()
+ default:
+ s.f.drainErr(errors.New("lateness is not supported"))
+ }
+ return s.f
+}
+
+type TumblingTimeWindows struct {
+ // internal state of the sliding time window
+ flow.ComponentState
+ // errorHandler is the error handler and set by the streamingFlow
+ errorHandler func(error)
+ // For TumblingTimeWindows
+ // Unit: Millisecond
+ size int64
+ // lateness is the maximum allowed lateness for incoming elements
+ // Unit: Millisecond
+ lateness int64
+
+ // guard snapshots
+ snapshotsMu sync.Mutex
+ snapshots map[int64]flow.AggregationOp
+ acc flow.AggregationOp
+
+ currentWatermark int64
+ // guard timerHeap
+ timerMu sync.Mutex
+ timerHeap *flow.DedupPriorityQueue
+
+ // aggregationFactory is the factory for creating aggregation operator
+ aggregationFactory flow.AggregationOpFactory
+
+ // For api.Operator
+ in chan flow.StreamRecord
+ out chan flow.StreamRecord
+ done chan struct{}
+ purgedWindow chan timeWindow
+}
+
+func (s *TumblingTimeWindows) In() chan<- flow.StreamRecord {
+ return s.in
+}
+
+func (s *TumblingTimeWindows) Out() <-chan flow.StreamRecord {
+ return s.out
+}
+
+func (s *TumblingTimeWindows) Setup(ctx context.Context) error {
+ if s.acc == nil {
+ s.acc = s.aggregationFactory()
+ }
+ // start processing
+ s.Add(1)
+ go s.receive()
+ // start emitting
+ s.Add(1)
+ go s.emit()
+
+ return nil
+}
+
+func (s *TumblingTimeWindows) emit() {
+ defer s.Done()
+ for w := range s.purgedWindow {
+ if err := func(window timeWindow) error {
+ s.snapshotsMu.Lock()
+ defer s.snapshotsMu.Unlock()
Review Comment:
Locking a channel writing is not good practice.
##########
pkg/flow/streaming/flow_sliding_window.go:
##########
@@ -0,0 +1,334 @@
+// 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 streaming
+
+import (
+ "container/heap"
+ "context"
+ "math"
+ "sync"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+)
+
+type TriggerResult bool
+
+const (
+ FIRE TriggerResult = true
+ CONTINUE = false
+)
+
+var (
+ _ flow.Operator = (*TumblingTimeWindows)(nil)
+ _ flow.WindowAssigner = (*TumblingTimeWindows)(nil)
+ _ flow.Window = (*timeWindow)(nil)
+
+ _ TriggerContext = (*triggerContext)(nil)
+)
+
+func (f *streamingFlow) Window(w flow.WindowAssigner) flow.WindowedFlow {
+ switch v := w.(type) {
+ case *TumblingTimeWindows:
+ v.errorHandler = f.drainErr
+ f.ops = append(f.ops, v)
+ default:
+ f.drainErr(errors.New("window type is not supported"))
+ }
+
+ return &windowedFlow{
+ f: f,
+ wa: w,
+ }
+}
+
+func (s *windowedFlow) AllowedLateness(lateness time.Duration) flow.Flow {
+ switch v := s.wa.(type) {
+ case *TumblingTimeWindows:
+ v.lateness = lateness.Milliseconds()
+ default:
+ s.f.drainErr(errors.New("lateness is not supported"))
+ }
+ return s.f
+}
+
+type TumblingTimeWindows struct {
+ // internal state of the sliding time window
+ flow.ComponentState
+ // errorHandler is the error handler and set by the streamingFlow
+ errorHandler func(error)
+ // For TumblingTimeWindows
+ // Unit: Millisecond
+ size int64
+ // lateness is the maximum allowed lateness for incoming elements
+ // Unit: Millisecond
+ lateness int64
+
+ // guard snapshots
+ snapshotsMu sync.Mutex
+ snapshots map[int64]flow.AggregationOp
+ acc flow.AggregationOp
+
+ currentWatermark int64
+ // guard timerHeap
+ timerMu sync.Mutex
+ timerHeap *flow.DedupPriorityQueue
+
+ // aggregationFactory is the factory for creating aggregation operator
+ aggregationFactory flow.AggregationOpFactory
+
+ // For api.Operator
+ in chan flow.StreamRecord
+ out chan flow.StreamRecord
+ done chan struct{}
+ purgedWindow chan timeWindow
Review Comment:
The sender and receiver are two routines. That means you don't have to use a channel to sync them. To merge them is more convince way.
##########
pkg/flow/streaming/flow_sliding_window.go:
##########
@@ -0,0 +1,334 @@
+// 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 streaming
+
+import (
+ "container/heap"
+ "context"
+ "math"
+ "sync"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+)
+
+type TriggerResult bool
+
+const (
+ FIRE TriggerResult = true
+ CONTINUE = false
+)
+
+var (
+ _ flow.Operator = (*TumblingTimeWindows)(nil)
+ _ flow.WindowAssigner = (*TumblingTimeWindows)(nil)
+ _ flow.Window = (*timeWindow)(nil)
+
+ _ TriggerContext = (*triggerContext)(nil)
+)
+
+func (f *streamingFlow) Window(w flow.WindowAssigner) flow.WindowedFlow {
+ switch v := w.(type) {
+ case *TumblingTimeWindows:
+ v.errorHandler = f.drainErr
+ f.ops = append(f.ops, v)
+ default:
+ f.drainErr(errors.New("window type is not supported"))
+ }
+
+ return &windowedFlow{
+ f: f,
+ wa: w,
+ }
+}
+
+func (s *windowedFlow) AllowedLateness(lateness time.Duration) flow.Flow {
+ switch v := s.wa.(type) {
+ case *TumblingTimeWindows:
+ v.lateness = lateness.Milliseconds()
+ default:
+ s.f.drainErr(errors.New("lateness is not supported"))
+ }
+ return s.f
+}
+
+type TumblingTimeWindows struct {
+ // internal state of the sliding time window
+ flow.ComponentState
+ // errorHandler is the error handler and set by the streamingFlow
+ errorHandler func(error)
+ // For TumblingTimeWindows
+ // Unit: Millisecond
+ size int64
+ // lateness is the maximum allowed lateness for incoming elements
+ // Unit: Millisecond
+ lateness int64
+
+ // guard snapshots
+ snapshotsMu sync.Mutex
+ snapshots map[int64]flow.AggregationOp
+ acc flow.AggregationOp
+
+ currentWatermark int64
+ // guard timerHeap
+ timerMu sync.Mutex
+ timerHeap *flow.DedupPriorityQueue
+
+ // aggregationFactory is the factory for creating aggregation operator
+ aggregationFactory flow.AggregationOpFactory
+
+ // For api.Operator
+ in chan flow.StreamRecord
+ out chan flow.StreamRecord
+ done chan struct{}
+ purgedWindow chan timeWindow
+}
+
+func (s *TumblingTimeWindows) In() chan<- flow.StreamRecord {
+ return s.in
+}
+
+func (s *TumblingTimeWindows) Out() <-chan flow.StreamRecord {
+ return s.out
+}
+
+func (s *TumblingTimeWindows) Setup(ctx context.Context) error {
+ if s.acc == nil {
+ s.acc = s.aggregationFactory()
+ }
+ // start processing
+ s.Add(1)
+ go s.receive()
+ // start emitting
+ s.Add(1)
+ go s.emit()
+
+ return nil
+}
+
+func (s *TumblingTimeWindows) emit() {
+ defer s.Done()
+ for w := range s.purgedWindow {
+ if err := func(window timeWindow) error {
+ s.snapshotsMu.Lock()
+ defer s.snapshotsMu.Unlock()
+
+ if snapshot, ok := s.snapshots[window.MaxTimestamp()]; ok {
+ if err := s.acc.Merge(snapshot); err != nil {
+ return err
+ }
+
+ s.out <- flow.NewStreamRecord(s.acc.Snapshot(), window.start)
+ }
+
+ return nil
+ }(w); err != nil {
+ s.errorHandler(err)
+ }
+ }
+ close(s.done)
+}
+
+func (s *TumblingTimeWindows) purgeOutdatedWindows() {
+ s.timerMu.Lock()
+ defer s.timerMu.Unlock()
+ for {
+ if lookAhead, ok := s.timerHeap.Peek().(*internalTimer); ok {
+ if lookAhead.triggerTimeMillis <= s.currentWatermark {
+ oldestTimer := heap.Pop(s.timerHeap).(*internalTimer)
+ s.purgedWindow <- oldestTimer.w
+ continue
+ }
+ }
+ return
+ }
+}
+
+func (s *TumblingTimeWindows) receive() {
+ defer s.Done()
+
+ for elem := range s.in {
+ assignedWindows, err := s.AssignWindows(elem.TimestampMillis())
+ if err != nil {
+ s.errorHandler(err)
+ continue
+ }
+ ctx := triggerContext{
+ delegation: s,
+ }
+ for _, w := range assignedWindows {
+ // drop if the window is late
+ if s.isWindowLate(w) {
+ continue
+ }
+ tw := w.(timeWindow)
+ ctx.window = tw
+ // add elem to the bucket
+ s.snapshotsMu.Lock()
+ if oldAggr, ok := s.snapshots[tw.MaxTimestamp()]; ok {
+ oldAggr.Add([]flow.StreamRecord{elem})
+ } else {
+ newAggr := s.aggregationFactory()
+ newAggr.Add([]flow.StreamRecord{elem})
+ s.snapshots[tw.MaxTimestamp()] = newAggr
+ }
+ s.snapshotsMu.Unlock()
+
+ result := ctx.OnElement(elem)
+ if result == FIRE {
+ s.purgedWindow <- tw
+ }
+ }
+
+ // even if the incoming elements do not follow strict order,
+ // the watermark could increase monotonically.
+ if elem.TimestampMillis() > s.currentWatermark {
+ s.currentWatermark = elem.TimestampMillis()
+
+ // TODO: add various strategies to allow lateness items, which come later than the current watermark
+ // Currently, assume the current watermark is t,
+ // then we allow lateness items coming after t-windowSize+slideSize, but before t,
+ // i.e. [t-windowSize+slideSize, t)
+ // NOTE: This call is the normal path to flush outdated windows immediately
+ // without considering lateness.
+ // The purged windows may be accumulated again due to non-zero lateness.
+ s.purgeOutdatedWindows()
+ }
+ }
+ close(s.purgedWindow)
+ <-s.done
+ close(s.out)
+}
+
+// isWindowLate checks whether this window is valid
+func (s *TumblingTimeWindows) isWindowLate(w flow.Window) bool {
+ return w.MaxTimestamp()+s.lateness <= s.currentWatermark
+}
+
+func (s *TumblingTimeWindows) Teardown(ctx context.Context) error {
+ s.Wait()
+ return nil
+}
+
+func (s *TumblingTimeWindows) Exec(downstream flow.Inlet) {
+ s.Add(1)
+ go flow.Transmit(&s.ComponentState, downstream, s)
+}
+
+func NewTumblingTimeWindows(size time.Duration) *TumblingTimeWindows {
+ return &TumblingTimeWindows{
+ size: size.Milliseconds(),
+ lateness: 0,
+ snapshots: make(map[int64]flow.AggregationOp),
+ timerHeap: flow.NewPriorityQueue(false),
+ in: make(chan flow.StreamRecord),
+ out: make(chan flow.StreamRecord),
+ done: make(chan struct{}),
+ purgedWindow: make(chan timeWindow),
+ currentWatermark: 0,
+ }
+}
+
+type timeWindow struct {
+ start int64
+ end int64
+}
+
+func (t timeWindow) MaxTimestamp() int64 {
+ return t.end - 1
+}
+
+// AssignWindows assigns windows according to the given timestamp
+func (s *TumblingTimeWindows) AssignWindows(timestamp int64) ([]flow.Window, error) {
+ if timestamp > math.MinInt64 {
+ start := getWindowStart(timestamp, s.size)
+ return []flow.Window{
+ timeWindow{
+ start: start,
+ end: start + s.size,
+ },
+ }, nil
+ }
+ return nil, errors.New("invalid timestamp from the element")
+}
+
+// getWindowStart calculates the window start for a timestamp.
+func getWindowStart(timestamp, windowSize int64) int64 {
+ remainder := timestamp % windowSize
+ return timestamp - remainder
+}
+
+// eventTimeTriggerOnElement processes element(s) with EventTimeTrigger
+func eventTimeTriggerOnElement(window timeWindow, ctx TriggerContext) TriggerResult {
+ if window.MaxTimestamp() <= ctx.GetCurrentWatermark() {
+ // if the watermark is already past the window fire immediately
+ return FIRE
+ }
+ ctx.RegisterEventTimeTimer(window.MaxTimestamp())
+ return CONTINUE
+}
+
+type TriggerContext interface {
Review Comment:
What's this exported interface used for? Its invokers are from this package. I don't find other implements more than triggerContext.
##########
pkg/flow/streaming/flow_topn.go:
##########
@@ -0,0 +1,171 @@
+// 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 streaming
+
+import (
+ "github.com/emirpasic/gods/maps/treemap"
+ "github.com/emirpasic/gods/utils"
+ "github.com/google/go-cmp/cmp"
+ "github.com/pkg/errors"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+)
+
+type windowedFlow struct {
+ f *streamingFlow
+ wa flow.WindowAssigner
+}
+
+func (s *windowedFlow) TopN(topNum int, opts ...any) flow.Flow {
+ s.wa.(*TumblingTimeWindows).aggregationFactory = func() flow.AggregationOp {
+ topNAggrFunc := &topNAggregator{
+ cacheSize: topNum,
+ sort: modelv1.Sort_SORT_UNSPECIFIED,
+ }
+ // apply user customized options
+ for _, opt := range opts {
+ if applier, ok := opt.(TopNOption); ok {
+ applier(topNAggrFunc)
+ }
+ }
+ if topNAggrFunc.sortKeyExtractor == nil {
+ s.f.drainErr(errors.New("sortKeyExtractor must be specified"))
+ }
+ if topNAggrFunc.sort == modelv1.Sort_SORT_DESC {
+ topNAggrFunc.comparator = func(a, b interface{}) int {
+ return utils.Int64Comparator(b, a)
+ }
+ } else {
+ topNAggrFunc.comparator = utils.Int64Comparator
+ }
Review Comment:
The order should default to DESC if users don't specific one.
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org