Add structured logging

cmd/send/send.go

@@ -5,6 +5,7 @@ import (
 	"hermetic/internal/common_flags"
 	"hermetic/internal/dps"
 	"hermetic/internal/teams"
+	"log/slog"
 
 	"github.com/spf13/cobra"
 )
@@ -59,7 +60,7 @@ func parseArgumentsAndCallSend(cmd *cobra.Command, args []string) error {
 		teamsErrorMessage := teams.CreateGeneralFailureMessage(err)
 		if err := teams.SendMessage(teamsErrorMessage, common_flags.TeamsWebhookNotificationUrl); err != nil {
 			err = fmt.Errorf("failed to send error message to Teams, cause: `%w`", err)
-			fmt.Printf("%s\n", err)
+			slog.Error(err.Error())
 		}
 		return err
 	}

cmd/verify/confirm/confirm.go

@@ -8,6 +8,7 @@ import (
 	"hermetic/internal/common_flags"
 	"hermetic/internal/teams"
 	confirmImplementation "hermetic/internal/verify/confirm"
+	"log/slog"
 	"os"
 	"os/signal"
 )
@@ -57,11 +58,11 @@ func parseArgumentsAndReadConfirmTopic(cmd *cobra.Command, args []string) error
 	err = confirmImplementation.ReadConfirmTopic(ctx, reader, confirmMessageReceiverUrl)
 	if err != nil {
 		err = fmt.Errorf("verification error, cause: `%w`", err)
-		fmt.Printf("Sending error message to Teams\n")
+		slog.Info("Sending error message to Teams")
 		teamsErrorMessage := teams.CreateGeneralFailureMessage(err)
 		if err := teams.SendMessage(teamsErrorMessage, common_flags.TeamsWebhookNotificationUrl); err != nil {
 			err = fmt.Errorf("failed to send error message to Teams, cause: `%w`", err)
-			fmt.Printf("%s\n", err)
+			slog.Error(err.Error())
 		}
 		return err
 	}

cmd/verify/reject/reject.go

@@ -8,6 +8,7 @@ import (
 	"hermetic/internal/common_flags"
 	"hermetic/internal/teams"
 	rejectImplementation "hermetic/internal/verify/reject"
+	"log/slog"
 	"os"
 	"os/signal"
 )
@@ -45,11 +46,11 @@ func parseArgumentsAndReadRejectTopic(cmd *cobra.Command, args []string) error {
 	err = rejectImplementation.ReadRejectTopic(ctx, reader, common_flags.TeamsWebhookNotificationUrl)
 	if err != nil {
 		err = fmt.Errorf("verification error, cause: `%w`", err)
-		fmt.Printf("Sending error message to Teams\n")
+		slog.Info("Sending error message to Teams")
 		teamsErrorMessage := teams.CreateGeneralFailureMessage(err)
 		if err := teams.SendMessage(teamsErrorMessage, common_flags.TeamsWebhookNotificationUrl); err != nil {
 			err = fmt.Errorf("failed to send error message to Teams, cause: `%w`", err)
-			fmt.Printf("%s\n", err)
+			slog.Error(err.Error())
 		}
 		return err
 	}

internal/dps/read.go

@@ -6,11 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"github.com/segmentio/kafka-go"
+	"log/slog"
 )
 
 func ReadMessages(ctx context.Context, reader *kafka.Reader) (*KafkaResponse, error) {
 	for {
-		fmt.Println("Reading next message...")
+		slog.Info("Reading next message...")
 		message, err := reader.ReadMessage(ctx)
 		if err != nil {
 			return nil, fmt.Errorf("failed to read message: %w", err)
@@ -22,14 +23,14 @@ func ReadMessages(ctx context.Context, reader *kafka.Reader) (*KafkaResponse, er
 		if err != nil {
 			syntaxError := new(json.SyntaxError)
 			if errors.As(err, &syntaxError) {
-				fmt.Printf("Could not read message at offset '%d', syntax error in message, skipping offset\n", message.Offset)
+				slog.Error("Could not read message at offset, syntax error in message, skipping offset", "offset", message.Offset)
 				continue
 			}
 			return nil, fmt.Errorf("failed to unmarshal json, original error: '%w'", err)
 		}
 
 		if !IsWebArchiveOwned(&dpsResponse) {
-			fmt.Printf("Message at offset '%d' is not owned by web archive, skipping offset\n", message.Offset)
+			slog.Info("Message at offset is not owned by web archive, skipping offset", "offset", message.Offset)
 			continue
 		}
 

internal/dps/send.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"net"
 	"os"
 	"time"
@@ -71,22 +72,22 @@ func readLatestMessages(kafkaEndpoints []string, transferTopicName string) ([]su
 	var messages []submission_information_package.Package
 
 	for offsetToReadFrom := offsets.first; offsetToReadFrom < offsets.last; offsetToReadFrom++ {
-		fmt.Printf("Reading message at offset '%d'\n", offsetToReadFrom)
+		slog.Info("Reading message ", "offset", offsetToReadFrom)
 		err := messageReader.Reader.SetOffset(offsetToReadFrom)
 		if err != nil {
 			return nil, fmt.Errorf("failed to set offset '%d', original error: '%w'", offsetToReadFrom, err)
 		}
 
 		message, err := messageReader.ReadMessageWithTimeout(readTimeout)
 		if errors.Is(err, context.DeadlineExceeded) {
-			fmt.Printf("Could not read message at offset '%d', read timeout '%s' exceeded, skipping offset\n", offsetToReadFrom, readTimeout)
+			slog.Error("Could not read message, timeout exceeded. Skipping offset.", "offset", offsetToReadFrom, "read_timeout", readTimeout)
 			continue
 		}
 		if err != nil {
 			return nil, fmt.Errorf("failed to read message at offset '%d', original error: '%w'", offsetToReadFrom, err)
 		}
 		if message.Value == nil {
-			fmt.Printf("Message at offset '%d' is nil, skipping offset\n", offsetToReadFrom)
+			slog.Info("Message at offset is nil, skipping offset", "offset", offsetToReadFrom)
 			continue
 		}
 		var submissionInformationPackage submission_information_package.Package
@@ -95,7 +96,7 @@ func readLatestMessages(kafkaEndpoints []string, transferTopicName string) ([]su
 		if err != nil {
 			syntaxError := new(json.SyntaxError)
 			if errors.As(err, &syntaxError) {
-				fmt.Printf("Could not read message at offset '%d', syntax error in message, skipping offset\n", offsetToReadFrom)
+				slog.Error("Could not read message at offset, syntax error in message, skipping offset", "offset", offsetToReadFrom)
 				continue
 			}
 			return nil, fmt.Errorf("failed to unmarshal json, original error: '%w'", err)
@@ -154,7 +155,7 @@ func PrepareAndSendSubmissionInformationPackage(kafkaEndpoints []string, transfe
 	}
 
 	for _, message := range relevantMessages {
-		fmt.Printf("Pushing '%s' to cache\n", message.Path)
+		slog.Error("Pushing message to cache", "path", message.Path)
 		err := cache.Set(message.Path, []byte("Sent"))
 		if err != nil {
 			return fmt.Errorf("failed to set '%s' in cache, original error: '%w'", message.Path, err)
@@ -171,7 +172,7 @@ func PrepareAndSendSubmissionInformationPackage(kafkaEndpoints []string, transfe
 			destinationPath := rootPath + directoryName
 			_, err := cache.Get(destinationPath)
 			if err == nil {
-				fmt.Printf("Skipping directory '%s' as it has already been processed.\n", destinationPath)
+				slog.Info("Skipping directory, already processed.", "path", destinationPath)
 				continue
 			} else {
 				if !errors.Is(err, bigcache.ErrEntryNotFound) {
@@ -181,7 +182,7 @@ func PrepareAndSendSubmissionInformationPackage(kafkaEndpoints []string, transfe
 			if !path.IsDir() {
 				return fmt.Errorf("found file '%s' in root path '%s', but expected only directories", path.Name(), rootPath)
 			}
-			fmt.Printf("Processing directory %s\n", destinationPath)
+			slog.Info("Processing directory", "path", destinationPath)
 			submissionInformationPackage := submission_information_package.CreatePackage(destinationPath, directoryName, warcContentType)
 
 			kafkaMessage, err := json.Marshal(submissionInformationPackage)

internal/kafka/sender.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"fmt"
+	"log/slog"
 
 	"github.com/google/uuid"
 	"github.com/segmentio/kafka-go"
@@ -14,7 +15,7 @@ type Sender struct {
 
 func (sender *Sender) SendMessageToKafkaTopic(payload []byte) error {
 	kafkaMessageUuid := uuid.New()
-	fmt.Printf("Sending message with uuid %s\n", kafkaMessageUuid)
+	slog.Info("Sending message to Kafka topic", "UUID", kafkaMessageUuid)
 	kafkaMessageUuidBytes, err := kafkaMessageUuid.MarshalText()
 	if err != nil {
 		return fmt.Errorf("failed to marshal uuid, original error: '%w'", err)

internal/teams/teams.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/carlmjohnson/requests"
@@ -36,7 +37,7 @@ type Message struct {
 func SendMessage(payload Message, webhookUrl string) error {
 	timeoutContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	fmt.Println("Sending message to Teams")
+	slog.Info("Sending message to Teams")
 	bytes, err := json.Marshal(payload)
 	if err != nil {
 		return fmt.Errorf("failed to marshal message, cause: `%w`", err)

internal/verify/confirm/confirm.go

@@ -16,6 +16,7 @@ import (
 func ReadConfirmTopic(ctx context.Context, reader *kafka.Reader, receiverUrl string) error {
 	for {
 		response, err := dps.ReadMessages(ctx, reader)
+		slog.Info("Received confirm message from DPS", "message", response.DPSResponse)
 		if err != nil {
 			return fmt.Errorf("failed to read message from confirm-topic: `%w`", err)
 		}
@@ -24,8 +25,6 @@ func ReadConfirmTopic(ctx context.Context, reader *kafka.Reader, receiverUrl str
 			if err != nil {
 				return fmt.Errorf("failed to send confirm message: `%w`", err)
 			}
-		} else {
-			fmt.Printf("Received message: %v\n", response.DPSResponse)
 		}
 	}
 }
@@ -61,6 +60,6 @@ func SendConfirmMessage(baseUrl string, response dps.DigitalPreservationSystemRe
 	if resp.StatusCode != http.StatusOK {
 		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
 	}
-	fmt.Printf("Successfully sent confirm message to %s\n", url)
+	slog.Info("Successfully sent confirm message", "receiver_url", url)
 	return nil
 }

internal/verify/reject/reject.go

@@ -6,6 +6,7 @@ import (
 	"github.com/segmentio/kafka-go"
 	"hermetic/internal/dps"
 	"hermetic/internal/teams"
+	"log/slog"
 )
 
 func ReadRejectTopic(ctx context.Context, reader *kafka.Reader, teamsWebhookNotificationUrl string) error {
@@ -21,7 +22,7 @@ func ReadRejectTopic(ctx context.Context, reader *kafka.Reader, teamsWebhookNoti
 }
 
 func ProcessMessagesFromRejectTopic(reader *kafka.Reader, response *dps.KafkaResponse, teamsWebhookNotificationUrl string) error {
-	fmt.Printf("Processing message with ContentCategory: '%s'\n", response.DPSResponse.ContentCategory)
+	slog.Info("Processing message from reject-topic", "content_category", response.DPSResponse.ContentCategory)
 	payload := createTeamsDigitalPreservationSystemFailureMessage(response, reader.Config().Topic, reader.Config().Brokers)
 	err := teams.SendMessage(payload, teamsWebhookNotificationUrl)
 	if err != nil {

main.go

@@ -1,14 +1,18 @@
 package main
 
 import (
-	"fmt"
 	"hermetic/cmd"
+	"log/slog"
 	"os"
 )
 
 func main() {
+	handler := slog.NewJSONHandler(os.Stdout, nil)
+	logger := slog.New(handler)
+	slog.SetDefault(logger)
+
 	if err := cmd.NewRootCommand().Execute(); err != nil {
-		fmt.Printf("failed to execute command, got error: '%s'\n", err)
+		slog.Error("failed to execute command, got error:", err)
 		os.Exit(1)
 	}
 }