From c2522ffb59514443816a96386c16bb7527cbe57c Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 21 Aug 2021 14:54:24 +0300 Subject: read files bytewise for more control of whats happening - change transport protocol for more control over newlines --- internal/mapr/aggregateset.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) (limited to 'internal/mapr/aggregateset.go') diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go index a6cc6eb..029d87b 100644 --- a/internal/mapr/aggregateset.go +++ b/internal/mapr/aggregateset.go @@ -5,6 +5,8 @@ import ( "fmt" "strconv" "strings" + + "github.com/mimecast/dtail/internal/protocol" ) // AggregateSet represents aggregated key/value pairs from the @@ -70,20 +72,20 @@ func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- var sb strings.Builder sb.WriteString(groupKey) - sb.WriteString("āž”") - sb.WriteString(fmt.Sprintf("%dāž”", s.Samples)) + sb.WriteString(protocol.AggregateDelimiter) + sb.WriteString(fmt.Sprintf("%d%s", s.Samples, protocol.AggregateDelimiter)) for k, v := range s.FValues { sb.WriteString(k) sb.WriteString("=") - sb.WriteString(fmt.Sprintf("%vāž”", v)) + sb.WriteString(fmt.Sprintf("%v%s", v, protocol.AggregateDelimiter)) } for k, v := range s.SValues { sb.WriteString(k) sb.WriteString("=") sb.WriteString(v) - sb.WriteString("āž”") + sb.WriteString(protocol.AggregateDelimiter) } select { -- cgit v1.2.3 From 16dc57e1e1c28e9d762424e596223a980770e059 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 8 Sep 2021 19:10:50 +0300 Subject: mapreduce tables are in colors now too --- internal/mapr/aggregateset.go | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) (limited to 'internal/mapr/aggregateset.go') diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go index 029d87b..47f4925 100644 --- a/internal/mapr/aggregateset.go +++ b/internal/mapr/aggregateset.go @@ -6,6 +6,8 @@ import ( "strconv" "strings" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/protocol" ) @@ -68,22 +70,25 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error { // Serialize the aggregate set so it can be sent over the wire. func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) { - //logger.Trace("Serialising mapr.AggregateSet", s) - var sb strings.Builder + logger.Trace("Serialising mapr.AggregateSet", s) + sb := pool.BuilderBuffer.Get().(*strings.Builder) + defer pool.RecycleBuilderBuffer(sb) sb.WriteString(groupKey) sb.WriteString(protocol.AggregateDelimiter) - sb.WriteString(fmt.Sprintf("%d%s", s.Samples, protocol.AggregateDelimiter)) + sb.WriteString(fmt.Sprintf("%d", s.Samples)) + sb.WriteString(protocol.AggregateDelimiter) for k, v := range s.FValues { sb.WriteString(k) - sb.WriteString("=") - sb.WriteString(fmt.Sprintf("%v%s", v, protocol.AggregateDelimiter)) + sb.WriteString(protocol.AggregateKVDelimiter) + sb.WriteString(fmt.Sprintf("%v", v)) + sb.WriteString(protocol.AggregateDelimiter) } for k, v := range s.SValues { sb.WriteString(k) - sb.WriteString("=") + sb.WriteString(protocol.AggregateKVDelimiter) sb.WriteString(v) sb.WriteString(protocol.AggregateDelimiter) } -- cgit v1.2.3 From fe3e68afd99d8ea246be52893730f987e138ec24 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 19 Sep 2021 13:22:59 +0300 Subject: move args to config package logger package rewrite as dlog --- internal/mapr/aggregateset.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'internal/mapr/aggregateset.go') diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go index 47f4925..14e6943 100644 --- a/internal/mapr/aggregateset.go +++ b/internal/mapr/aggregateset.go @@ -6,7 +6,7 @@ import ( "strconv" "strings" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/protocol" ) @@ -37,7 +37,7 @@ func (s *AggregateSet) String() string { // Merge one aggregate set into this one. func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error { s.Samples += set.Samples - //logger.Trace("Merge", set) + //dlog.Common.Trace("Merge", set) for _, sc := range query.Select { storage := sc.FieldStorage @@ -70,7 +70,7 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error { // Serialize the aggregate set so it can be sent over the wire. func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) { - logger.Trace("Serialising mapr.AggregateSet", s) + dlog.Common.Trace("Serialising mapr.AggregateSet", s) sb := pool.BuilderBuffer.Get().(*strings.Builder) defer pool.RecycleBuilderBuffer(sb) -- cgit v1.2.3 From 97747ea0f3178f7f5890512d483fdccaa82846b0 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 9 Oct 2021 21:10:29 +0300 Subject: vetting and linting and some code restyling --- internal/mapr/aggregateset.go | 4 ---- 1 file changed, 4 deletions(-) (limited to 'internal/mapr/aggregateset.go') diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go index 14e6943..c50c7a1 100644 --- a/internal/mapr/aggregateset.go +++ b/internal/mapr/aggregateset.go @@ -38,7 +38,6 @@ func (s *AggregateSet) String() string { func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error { s.Samples += set.Samples //dlog.Common.Trace("Merge", set) - for _, sc := range query.Select { storage := sc.FieldStorage switch sc.Operation { @@ -115,7 +114,6 @@ func (s *AggregateSet) addFloatMin(key string, value float64) { s.FValues[key] = value return } - if f > value { s.FValues[key] = value } @@ -128,7 +126,6 @@ func (s *AggregateSet) addFloatMax(key string, value float64) { s.FValues[key] = value return } - if f < value { s.FValues[key] = value } @@ -147,7 +144,6 @@ func (s *AggregateSet) setFloat(key string, value float64) { // Aggregate data to the aggregate set. func (s *AggregateSet) Aggregate(key string, agg AggregateOperation, value string, clientAggregation bool) (err error) { var f float64 - // First check if we can aggregate anything without converting value to float. switch agg { case Count: -- cgit v1.2.3