v1.1.0 counters and mark open telemetry span as good if all passed

server.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -164,6 +165,8 @@ type Server struct {
 	bytesRead                uint64
 	bytesWritten             uint64
 	transactionsAll          uint64
+	transactionsSuccess      uint64
+	transactionsFail         uint64
 	transactionsActive       int32
 	lastTransactionStartedAt time.Time
 }
@@ -278,6 +281,7 @@ func (srv *Server) ListenAndServe(addr string) error {
 func (srv *Server) runCloseHandlers(transaction *Transaction) {
 	transaction.mu.Lock()
 	defer transaction.mu.Unlock()
+	closedProperly := true
 	if transaction.closeHandlersCalled {
 		transaction.LogDebug("close handlers already called")
 		return
@@ -288,6 +292,7 @@ func (srv *Server) runCloseHandlers(transaction *Transaction) {
 		srv.Logger.Debugf(transaction, "Starting close handler %v...", k)
 		closeError = srv.CloseHandlers[k](transaction)
 		if closeError != nil {
+			closedProperly = false
 			transaction.LogError(closeError, "while calling close handler")
 		} else {
 			transaction.LogDebug("closing handler %v is called", k)
@@ -296,6 +301,16 @@ func (srv *Server) runCloseHandlers(transaction *Transaction) {
 	transaction.closeHandlersCalled = true
 	srv.Logger.Infof(transaction, "Closing transaction %s.", transaction.ID)
 	atomic.AddInt32(&srv.transactionsActive, -1)
+	if closedProperly {
+		if transaction.dataHandlersCalledProperly {
+			transaction.Span.SetStatus(codes.Ok, "Transaction completed")
+			atomic.AddUint64(&srv.transactionsSuccess, 1)
+		} else {
+			atomic.AddUint64(&srv.transactionsFail, 1)
+		}
+	} else {
+		atomic.AddUint64(&srv.transactionsFail, 1)
+	}
 }
 
 // Serve starts the SMTP server and listens on the Listener provided

server_counters.go

@@ -52,6 +52,16 @@ func (srv *Server) GetActiveTransactionsCount() int32 {
 	return srv.transactionsActive
 }
 
+// GetSuccessfulTransactionsCount returns number of successful transactions this server processed
+func (srv *Server) GetSuccessfulTransactionsCount() uint64 {
+	return srv.transactionsSuccess
+}
+
+// GetFailedTransactionsCount returns number of failed transactions this server processed
+func (srv *Server) GetFailedTransactionsCount() uint64 {
+	return srv.transactionsFail
+}
+
 // ResetCounters resets counters
 func (srv *Server) ResetCounters() {
 	srv.bytesRead = 0
@@ -73,6 +83,11 @@ func (srv *Server) StartPrometheusScrapperEndpoint(address, path string) (err er
 			srv.Hostname, srv.GetActiveTransactionsCount(), srv.lastTransactionStartedAt.UnixMilli())
 		fmt.Fprintf(res, "all_transactions_count{hostname=\"%s\"} %v %v\n",
 			srv.Hostname, srv.GetTransactionsCount(), srv.lastTransactionStartedAt.UnixMilli())
+		fmt.Fprintf(res, "successfull_transactions_count{hostname=\"%s\"} %v %v\n",
+			srv.Hostname, srv.GetSuccessfulTransactionsCount(), srv.lastTransactionStartedAt.UnixMilli())
+		fmt.Fprintf(res, "failed_transactions_count{hostname=\"%s\"} %v %v\n",
+			srv.Hostname, srv.GetFailedTransactionsCount(), srv.lastTransactionStartedAt.UnixMilli())
+
 	})
 	return http.ListenAndServe(address, http.DefaultServeMux)
 }

transaction.go

@@ -95,7 +95,11 @@ type Transaction struct {
 	writer  *bufio.Writer
 	scanner *bufio.Scanner
 
+	// closeHandlersCalled used to ensure close handlers are called only once
 	closeHandlersCalled bool
+	// dataHandlersCalledProperly shows if data handlers for transaction are called properly,
+	// so we consider it is delivered
+	dataHandlersCalledProperly bool
 }
 
 // Context returns transaction context, which is canceled when transaction is closed

transaction_data.go

@@ -187,6 +187,7 @@ func (t *Transaction) handleDATA(cmd command) {
 			t.reply(250, "Thank you.")
 			t.Love(commandExecutedProperly)
 			t.reset()
+			t.dataHandlersCalledProperly = true
 			return
 		}
 		t.LogError(err, "possible network error while reading message data")