only emit logs in retry writer

Author: corporatemax

src/pkg/egress/syslog/retry_writer.go

@@ -1,7 +1,6 @@
 package syslog
 
 import (
-	"fmt"
 	"log"
 	"math"
 	"time"
@@ -22,22 +21,19 @@ type RetryWriter struct {
 	retryDuration RetryDuration
 	maxRetries    int
 	binding       *URLBinding
-	emitter       AppLogEmitter
 }
 
 func NewRetryWriter(
 	urlBinding *URLBinding,
 	retryDuration RetryDuration,
 	maxRetries int,
 	writer egress.WriteCloser,
-	emitter AppLogEmitter,
 ) (egress.WriteCloser, error) {
 	return &RetryWriter{
 		Writer:        writer,
 		retryDuration: retryDuration,
 		maxRetries:    maxRetries,
 		binding:       urlBinding,
-		emitter:       emitter,
 	}, nil
 }
 
@@ -59,7 +55,6 @@ func (r *RetryWriter) Write(e *loggregator_v2.Envelope) error {
 
 		sleepDuration := r.retryDuration(i)
 		log.Printf(logTemplate, r.binding.URL.Host, sleepDuration, err)
-		r.emitter.EmitLog(e.SourceId, fmt.Sprintf(logTemplate, r.binding.URL.Host, sleepDuration, err))
 
 		time.Sleep(sleepDuration)
 	}

src/pkg/egress/syslog/retry_writer_test.go

@@ -1,7 +1,6 @@
 package syslog_test
 
 import (
-	"code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper"
 	"errors"
 	"net/url"
 	"sync/atomic"
@@ -176,13 +175,10 @@ func buildRetryWriter(
 	maxRetries int,
 	delayMultiplier time.Duration,
 ) (egress.WriteCloser, error) {
-	factory := syslog.NewAppLogEmitterFactory()
-	emitter := factory.NewAppLogEmitter(testhelper.NewSpyLogClient(), "test-index")
 	return syslog.NewRetryWriter(
 		urlBinding,
 		syslog.RetryDuration(buildDelay(delayMultiplier)),
 		maxRetries,
 		w,
-		emitter,
 	)
 }

src/pkg/egress/syslog/syslog_connector.go

@@ -122,8 +122,8 @@ func (w *SyslogConnector) Connect(ctx context.Context, b Binding) (egress.Writer
 		w.droppedMetric.Add(float64(missed))
 		drainDroppedMetric.Add(float64(missed))
 
-		w.appLogEmitter.EmitLog(b.AppId, fmt.Sprintf("%d messages lost for application %s in user provided syslog drain with url %s", missed, b.AppId, anonymousUrl.String()))
 		w.emitStandardOutErrorLog(b.AppId, urlBinding.Scheme(), anonymousUrl.String(), missed)
+		w.appLogEmitter.EmitLog(b.AppId, fmt.Sprintf("%d messages lost for application %s in user provided syslog drain with url %s", missed, b.AppId, anonymousUrl.String()))
 	}), w.wg)
 
 	filteredWriter, err := NewFilteringDrainWriter(b, dw)

src/pkg/egress/syslog/tcp.go

@@ -32,6 +32,8 @@ type TCPWriter struct {
 	syslogConverter *Converter
 
 	egressMetric metrics.Counter
+
+	emitter AppLogEmitter
 }
 
 // NewTCPWriter creates a new TCP syslog writer.
@@ -40,6 +42,7 @@ func NewTCPWriter(
 	netConf NetworkTimeoutConfig,
 	egressMetric metrics.Counter,
 	c *Converter,
+	emitter AppLogEmitter,
 ) egress.WriteCloser {
 	dialer := &net.Dialer{
 		Timeout:   netConf.DialTimeout,
@@ -58,6 +61,7 @@ func NewTCPWriter(
 		scheme:          "syslog",
 		egressMetric:    egressMetric,
 		syslogConverter: c,
+		emitter:         emitter,
 	}
 
 	return w
@@ -104,7 +108,8 @@ func (w *TCPWriter) connection() (net.Conn, error) {
 func (w *TCPWriter) connect() (net.Conn, error) {
 	conn, err := w.dialFunc(w.url.Host)
 	if err != nil {
-		// todo send logs to app logstream
+		logMessage := fmt.Sprintf("Failed to connect to %s", w.url.String())
+		w.emitter.EmitLog(w.appID, logMessage)
 		return nil, err
 	}
 	w.conn = conn

src/pkg/egress/syslog/tcp_test.go

@@ -2,6 +2,7 @@ package syslog_test
 
 import (
 	"bufio"
+	"code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper"
 	"fmt"
 	"io"
 	"net"
@@ -51,12 +52,16 @@ var _ = Describe("TCPWriter", func() {
 		BeforeEach(func() {
 			var err error
 			egressCounter = &metricsHelpers.SpyMetric{}
+			factory := syslog.NewAppLogEmitterFactory()
+			logClient := testhelper.NewSpyLogClient()
+			emitter := factory.NewAppLogEmitter(logClient, "3")
 
 			writer = syslog.NewTCPWriter(
 				binding,
 				netConf,
 				egressCounter,
 				syslog.NewConverter(),
+				emitter,
 			)
 			Expect(err).ToNot(HaveOccurred())
 		})
@@ -183,12 +188,16 @@ var _ = Describe("TCPWriter", func() {
 		It("write returns an error", func() {
 			env := buildLogEnvelope("APP", "2", "just a test", loggregator_v2.Log_OUT)
 			binding.URL, _ = url.Parse("syslog://localhost-garbage:9999")
+			factory := syslog.NewAppLogEmitterFactory()
+			logClient := testhelper.NewSpyLogClient()
+			emitter := factory.NewAppLogEmitter(logClient, "3")
 
 			writer := syslog.NewTCPWriter(
 				binding,
 				netConf,
 				&metricsHelpers.SpyMetric{},
 				syslog.NewConverter(),
+				emitter,
 			)
 
 			errs := make(chan error, 1)
@@ -208,11 +217,15 @@ var _ = Describe("TCPWriter", func() {
 		Context("with a happy dialer", func() {
 			BeforeEach(func() {
 				var err error
+				factory := syslog.NewAppLogEmitterFactory()
+				logClient := testhelper.NewSpyLogClient()
+				emitter := factory.NewAppLogEmitter(logClient, "3")
 				writer = syslog.NewTCPWriter(
 					binding,
 					netConf,
 					&metricsHelpers.SpyMetric{},
 					syslog.NewConverter(),
+					emitter,
 				)
 				Expect(err).ToNot(HaveOccurred())
 

src/pkg/egress/syslog/tls.go

@@ -27,6 +27,7 @@ func NewTLSWriter(
 	tlsConf *tls.Config,
 	egressMetric metrics.Counter,
 	syslogConverter *Converter,
+	emitter AppLogEmitter,
 ) egress.WriteCloser {
 
 	dialer := &net.Dialer{
@@ -48,6 +49,7 @@ func NewTLSWriter(
 			scheme:          "syslog-tls",
 			egressMetric:    egressMetric,
 			syslogConverter: syslogConverter,
+			emitter:         emitter,
 		},
 	}
 

src/pkg/egress/syslog/tls_test.go

@@ -55,6 +55,9 @@ var _ = Describe("TLSWriter", func() {
 			Hostname: "test-hostname",
 			URL:      url,
 		}
+		factory := syslog.NewAppLogEmitterFactory()
+		logClient := testhelper.NewSpyLogClient()
+		emitter := factory.NewAppLogEmitter(logClient, "3")
 		writer := syslog.NewTLSWriter(
 			binding,
 			netConf,
@@ -63,6 +66,7 @@ var _ = Describe("TLSWriter", func() {
 			},
 			egressCounter,
 			syslog.NewConverter(),
+			emitter,
 		)
 		defer writer.Close()
 

src/pkg/egress/syslog/writer_factory.go

@@ -62,7 +62,6 @@ func (f WriterFactory) NewWriter(ub *URLBinding, emitter AppLogEmitter) (egress.
 		if err != nil {
 			errorMessage := err.Error()
 			err = NewWriterFactoryErrorf(ub.URL, "failed to load certificate: %s", errorMessage)
-			emitter.EmitLog(ub.AppID, fmt.Sprintf("failed to load certificate: %s", errorMessage))
 			return nil, err
 		}
 		tlsCfg.Certificates = []tls.Certificate{cert}
@@ -71,7 +70,6 @@ func (f WriterFactory) NewWriter(ub *URLBinding, emitter AppLogEmitter) (egress.
 		ok := tlsCfg.RootCAs.AppendCertsFromPEM(ub.CA)
 		if !ok {
 			err := NewWriterFactoryErrorf(ub.URL, "failed to load root CA")
-			emitter.EmitLog(ub.AppID, "failed to load root CA")
 			return nil, err
 		}
 	}
@@ -123,6 +121,7 @@ func (f WriterFactory) NewWriter(ub *URLBinding, emitter AppLogEmitter) (egress.
 			f.netConf,
 			egressMetric,
 			converter,
+			emitter,
 		)
 	case "syslog-tls":
 		w = NewTLSWriter(
@@ -131,6 +130,7 @@ func (f WriterFactory) NewWriter(ub *URLBinding, emitter AppLogEmitter) (egress.
 			tlsCfg,
 			egressMetric,
 			converter,
+			emitter,
 		)
 	}
 
@@ -143,6 +143,5 @@ func (f WriterFactory) NewWriter(ub *URLBinding, emitter AppLogEmitter) (egress.
 		ExponentialDuration,
 		maxRetries,
 		w,
-		emitter,
 	)
 }