Locking cwDest on Publish. Remove vestigial functionality

Author: dricross

plugins/outputs/cloudwatchlogs/cloudwatchlogs.go

@@ -4,7 +4,6 @@
 package cloudwatchlogs
 
 import (
-	"encoding/json"
 	"fmt"
 	"regexp"
 	"strings"
@@ -28,7 +27,6 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/logs"
 	"github.com/aws/amazon-cloudwatch-agent/plugins/outputs/cloudwatchlogs/internal/pusher"
 	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
-	"github.com/aws/amazon-cloudwatch-agent/tool/util"
 )
 
 const (
@@ -40,10 +38,7 @@ const (
 
 	defaultFlushTimeout = 5 * time.Second
 
-	maxRetryTimeout    = 14*24*time.Hour + 10*time.Minute
-	metricRetryTimeout = 2 * time.Minute
-
-	attributesInFields = "attributesInFields"
+	maxRetryTimeout = 14*24*time.Hour + 10*time.Minute
 )
 
 var (
@@ -95,7 +90,7 @@ func (c *CloudWatchLogs) Close() error {
 
 	c.cwDests.Range(func(_, value interface{}) bool {
 		if d, ok := value.(*cwDest); ok {
-			d.stop()
+			d.Stop()
 		}
 		return true
 	})
@@ -110,10 +105,8 @@ func (c *CloudWatchLogs) Close() error {
 }
 
 func (c *CloudWatchLogs) Write(metrics []telegraf.Metric) error {
-	for _, m := range metrics {
-		c.writeMetricAsStructuredLog(m)
-	}
-	return nil
+	// we no longer expect this to be used. We now use the OTel awsemfexporter for sending EMF metrics to CloudWatch Logs
+	return fmt.Errorf("unexpected call to Write")
 }
 
 func (c *CloudWatchLogs) CreateDest(group, stream string, retention int, logGroupClass string, logSrc logs.LogSrc) logs.LogDest {
@@ -161,9 +154,14 @@ func (c *CloudWatchLogs) getDest(t pusher.Target, logSrc logs.LogSrc) *cwDest {
 		c.targetManager = pusher.NewTargetManager(c.Log, client)
 	})
 	p := pusher.NewPusher(c.Log, t, client, c.targetManager, logSrc, c.workerPool, c.ForceFlushInterval.Duration, maxRetryTimeout, &c.pusherWaitGroup)
-	cwd := &cwDest{pusher: p, retryer: logThrottleRetryer, onStopFunc: func() {
-		c.cwDests.Delete(t)
-	}}
+	cwd := &cwDest{
+		pusher:   p,
+		retryer:  logThrottleRetryer,
+		refCount: 1,
+		onStopFunc: func() {
+			c.cwDests.Delete(t)
+		},
+	}
 	c.cwDests.Store(t, cwd)
 	return cwd
 }
@@ -201,136 +199,44 @@ func (c *CloudWatchLogs) createClient(retryer aws.RequestRetryer) *cloudwatchlog
 	return client
 }
 
-func (c *CloudWatchLogs) writeMetricAsStructuredLog(m telegraf.Metric) {
-	t, err := c.getTargetFromMetric(m)
-	if err != nil {
-		c.Log.Errorf("Failed to find target: %v", err)
-	}
-	cwd := c.getDest(t, nil)
-	if cwd == nil {
-		c.Log.Warnf("unable to find log destination, group: %v, stream: %v", t.Group, t.Stream)
-		return
-	}
-	cwd.switchToEMF()
-	cwd.pusher.Sender.SetRetryDuration(metricRetryTimeout)
-
-	e := c.getLogEventFromMetric(m)
-	if e == nil {
-		return
-	}
-
-	cwd.AddEvent(e)
-}
-
-func (c *CloudWatchLogs) getTargetFromMetric(m telegraf.Metric) (pusher.Target, error) {
-	tags := m.Tags()
-	logGroup, ok := tags[LogGroupNameTag]
-	if !ok {
-		return pusher.Target{}, fmt.Errorf("structuredlog receive a metric with name '%v' without log group name", m.Name())
-	} else {
-		m.RemoveTag(LogGroupNameTag)
-	}
-
-	logStream, ok := tags[LogStreamNameTag]
-	if ok {
-		m.RemoveTag(LogStreamNameTag)
-	} else if logStream == "" {
-		logStream = c.LogStreamName
-	}
-
-	return pusher.Target{Group: logGroup, Stream: logStream, Class: util.StandardLogGroupClass, Retention: -1}, nil
+// Description returns a one-sentence description on the Output
+func (c *CloudWatchLogs) Description() string {
+	return "Configuration for AWS CloudWatchLogs output."
 }
 
-func (c *CloudWatchLogs) getLogEventFromMetric(metric telegraf.Metric) *structuredLogEvent {
-	var message string
-	if metric.HasField(LogEntryField) {
-		var ok bool
-		if message, ok = metric.Fields()[LogEntryField].(string); !ok {
-			c.Log.Warnf("The log entry value field is not string type: %v", metric.Fields())
-			return nil
-		}
-	} else {
-		content := map[string]interface{}{}
-		tags := metric.Tags()
-		// build all the attributesInFields
-		if val, ok := tags[attributesInFields]; ok {
-			attributes := strings.Split(val, ",")
-			mFields := metric.Fields()
-			for _, attr := range attributes {
-				if fieldVal, ok := mFields[attr]; ok {
-					content[attr] = fieldVal
-					metric.RemoveField(attr)
-				}
-			}
-			metric.RemoveTag(attributesInFields)
-			delete(tags, attributesInFields)
-		}
-
-		// build remaining attributes
-		for k := range tags {
-			content[k] = tags[k]
-		}
-
-		for k, v := range metric.Fields() {
-			var value interface{}
-
-			switch t := v.(type) {
-			case int:
-				value = float64(t)
-			case int32:
-				value = float64(t)
-			case int64:
-				value = float64(t)
-			case uint:
-				value = float64(t)
-			case uint32:
-				value = float64(t)
-			case uint64:
-				value = float64(t)
-			case float64:
-				value = t
-			case bool:
-				value = t
-			case string:
-				value = t
-			case time.Time:
-				value = float64(t.Unix())
-
-			default:
-				c.Log.Errorf("Detected unexpected fields (%s,%v) when encoding structured log event, value type %T is not supported", k, v, v)
-				return nil
-			}
-			content[k] = value
-		}
-
-		jsonMap, err := json.Marshal(content)
-		if err != nil {
-			c.Log.Errorf("Unalbe to marshal structured log content: %v", err)
-		}
-		message = string(jsonMap)
-	}
-
-	return &structuredLogEvent{
-		msg: message,
-		t:   metric.Time(),
-	}
-}
+var sampleConfig = `
+  ## Amazon REGION
+  region = "us-east-1"
 
-type structuredLogEvent struct {
-	msg string
-	t   time.Time
-}
+  ## Amazon Credentials
+  ## Credentials are loaded in the following order
+  ## 1) Assumed credentials via STS if role_arn is specified
+  ## 2) explicit credentials from 'access_key' and 'secret_key'
+  ## 3) shared profile from 'profile'
+  ## 4) environment variables
+  ## 5) shared credentials file
+  ## 6) EC2 Instance Profile
+  #access_key = ""
+  #secret_key = ""
+  #token = ""
+  #role_arn = ""
+  #profile = ""
+  #shared_credential_file = ""
 
-func (e *structuredLogEvent) Message() string {
-	return e.msg
-}
+  # The log stream name.
+  log_stream_name = "<log_stream_name>"
+`
 
-func (e *structuredLogEvent) Time() time.Time {
-	return e.t
+// SampleConfig returns the default configuration of the Output
+func (c *CloudWatchLogs) SampleConfig() string {
+	return sampleConfig
 }
 
-func (e *structuredLogEvent) Done() {}
-
+// cwDest is responsible for publishing logs from log files to a log group + log stream.
+// Logs from more than one log file may be published to the same destination. cwDest closes
+// itself when all log file tailers which referenced this cwDest are closed.
+// All exported functions should practice thread-safety by acquiring lock the cwDest
+// and not calling any other function which requires the lock.
 type cwDest struct {
 	pusher *pusher.Pusher
 	sync.Mutex
@@ -349,8 +255,10 @@ type cwDest struct {
 var _ logs.LogDest = (*cwDest)(nil)
 
 func (cd *cwDest) Publish(events []logs.LogEvent) error {
+	cd.Lock()
+	defer cd.Unlock()
 	if cd.stopped {
-		return fmt.Errorf("cannot publish events: destination has been stopped")
+		return logs.ErrOutputStopped
 	}
 	for _, e := range events {
 		if !cd.isEMF {
@@ -359,10 +267,7 @@ func (cd *cwDest) Publish(events []logs.LogEvent) error {
 				cd.switchToEMF()
 			}
 		}
-		cd.AddEvent(e)
-	}
-	if cd.stopped {
-		return logs.ErrOutputStopped
+		cd.addEvent(e)
 	}
 	return nil
 }
@@ -376,6 +281,12 @@ func (cd *cwDest) NotifySourceStopped() {
 	}
 }
 
+func (cd *cwDest) Stop() {
+	cd.Lock()
+	defer cd.Unlock()
+	cd.stop()
+}
+
 func (cd *cwDest) stop() {
 	if cd.stopped {
 		return
@@ -388,11 +299,7 @@ func (cd *cwDest) stop() {
 	}
 }
 
-func (cd *cwDest) AddEvent(e logs.LogEvent) {
-	if cd.stopped {
-		// cannot add event, destination has been stopped
-		return
-	}
+func (cd *cwDest) addEvent(e logs.LogEvent) {
 	// Drop events for metric path logs when queue is full
 	if cd.isEMF {
 		cd.pusher.AddEventNonBlocking(e)
@@ -402,8 +309,6 @@ func (cd *cwDest) AddEvent(e logs.LogEvent) {
 }
 
 func (cd *cwDest) switchToEMF() {
-	cd.Lock()
-	defer cd.Unlock()
 	if !cd.isEMF {
 		cd.isEMF = true
 		cwl, ok := cd.pusher.Service.(*cloudwatchlogs.CloudWatchLogs)
@@ -413,39 +318,6 @@ func (cd *cwDest) switchToEMF() {
 	}
 }
 
-// Description returns a one-sentence description on the Output
-func (c *CloudWatchLogs) Description() string {
-	return "Configuration for AWS CloudWatchLogs output."
-}
-
-var sampleConfig = `
-  ## Amazon REGION
-  region = "us-east-1"
-
-  ## Amazon Credentials
-  ## Credentials are loaded in the following order
-  ## 1) Assumed credentials via STS if role_arn is specified
-  ## 2) explicit credentials from 'access_key' and 'secret_key'
-  ## 3) shared profile from 'profile'
-  ## 4) environment variables
-  ## 5) shared credentials file
-  ## 6) EC2 Instance Profile
-  #access_key = ""
-  #secret_key = ""
-  #token = ""
-  #role_arn = ""
-  #profile = ""
-  #shared_credential_file = ""
-
-  # The log stream name.
-  log_stream_name = "<log_stream_name>"
-`
-
-// SampleConfig returns the default configuration of the Output
-func (c *CloudWatchLogs) SampleConfig() string {
-	return sampleConfig
-}
-
 func init() {
 	outputs.Add("cloudwatchlogs", func() telegraf.Output {
 		return &CloudWatchLogs{