diff --git a/include/interceptors/log_sink.hpp b/include/interceptors/log_sink.hpp
new file mode 100644
index 0000000..c0bd317
--- /dev/null
+++ b/include/interceptors/log_sink.hpp
@@ -0,0 +1,50 @@
+#pragma once
+#include <absl/log/log_sink.h>
+#include <absl/log/log_sink_registry.h>
+#include <absl/log/log_entry.h>
+#include <absl/log/initialize.h>
+#include <absl/log/globals.h>
+#include <iostream>
+
+namespace moderation::logging {
+
+class ReadableLogSink : public absl::LogSink {
+public:
+    std::atomic<bool> installed{false};
+
+    void Send(const absl::LogEntry& entry) override {
+        if (!installed) return; 
+        
+        std::string_view level;
+        switch (entry.log_severity()) {
+            case absl::LogSeverity::kInfo:    level = "INFO";    break;
+            case absl::LogSeverity::kWarning: level = "WARNING"; break;
+            case absl::LogSeverity::kError:   level = "ERROR";   break;
+            case absl::LogSeverity::kFatal:   level = "FATAL";   break;
+            default:                          level = "UNKNOWN";  break;
+        }
+        std::cerr
+            << "[" << level << "] "
+            << entry.source_basename() << ":" << entry.source_line() << " "
+            << entry.text_message() << "\n";
+    }
+
+    void Flush() override { std::cerr.flush(); }
+};
+
+inline ReadableLogSink* GetSink() {
+    static ReadableLogSink sink;
+    return &sink;
+}
+
+inline void InstallReadableSink() {
+    absl::AddLogSink(GetSink()); 
+}
+
+inline void ActivateReadableSink() {
+    absl::InitializeLog();
+    absl::SetStderrThreshold(absl::LogSeverityAtLeast::kInfinity);
+    GetSink()->installed = true;
+}
+
+} // namespace moderation::logging
\ No newline at end of file
diff --git a/include/interceptors/logger.hpp b/include/interceptors/logger.hpp
index 0235b0b..691772d 100644
--- a/include/interceptors/logger.hpp
+++ b/include/interceptors/logger.hpp
@@ -1,30 +1,190 @@
 #pragma once
 
-#include <grpcpp/grpcpp.h>
-#include <grpcpp/impl/codegen/server_interceptor.h>
-#include <grpcpp/server_context.h>
-#include <grpcpp/support/interceptor.h>
-#include <grpcpp/support/server_interceptor.h>
-#include <iostream>
+#include <absl/log/log.h>
+#include <absl/strings/ascii.h>
+#include <absl/strings/str_cat.h>
+#include <absl/strings/string_view.h>
+#include <absl/time/clock.h>
+#include <absl/time/time.h>
+
+#include <grpcpp/support/status.h>
+#include <librdkafka/rdkafkacpp.h>
+#include <unicode/utypes.h>
 #include <string>
 
-class LoggerInterceptor final : public grpc::experimental::Interceptor {
-  public:
-    explicit LoggerInterceptor(grpc::experimental::ServerRpcInfo* info) {
-        std::string method = info->method();
-        if (method == "unknown") {
-            std::cout << "[LoggerInterceptor] Unknown method called.\n";
-        } else {
-            std::cout << "[LoggerInterceptor] Method called: " << method << "\n";
-        }
-    }
-    void Intercept(grpc::experimental::InterceptorBatchMethods* methods) { methods->Proceed(); }
+namespace moderation::logging {
+
+enum class Subsystem {
+    kGrpc,
+    kKafka,
+    kIcu,
+    kAlgorithm,
+    kRepository,
+    kService,
+    kServer,
+    kConfig,
+    kUnknown
 };
 
-class LoggerInterceptorFactory : public grpc::experimental::ServerInterceptorFactoryInterface {
-  public:
-    grpc::experimental::Interceptor*
-    CreateServerInterceptor(grpc::experimental::ServerRpcInfo* info) {
-        return new LoggerInterceptor(info);
+inline absl::TimeZone GetLocalTZ() {
+    static absl::TimeZone timezone = [] {
+        absl::TimeZone time;
+        absl::LoadTimeZone("Europe/Berlin", &time);
+        return time;
+    }();
+    return timezone;
+}
+
+inline absl::string_view SubsystemName(Subsystem systemName) {
+    switch (systemName) {
+        case Subsystem::kGrpc: return "grpc";
+        case Subsystem::kKafka: return "kafka";
+        case Subsystem::kIcu: return "icu";
+        case Subsystem::kAlgorithm: return "algorithm";
+        case Subsystem::kRepository: return "repository";
+        case Subsystem::kService: return "service";
+        case Subsystem::kServer: return "server";
+        case Subsystem::kConfig: return "config";
+        default: return "unknown";
     }
-};
+}
+
+// --- explanation mapping per subsystem ---
+
+inline absl::string_view Explain(Subsystem, grpc::StatusCode code) {
+    switch (code) {
+        case grpc::StatusCode::DEADLINE_EXCEEDED:
+            return "client or server timeout, check network or increase deadline";
+        case grpc::StatusCode::UNAVAILABLE:
+            return "service unavailable, check endpoint health and connectivity";
+        case grpc::StatusCode::INVALID_ARGUMENT:
+            return "request validation failed, verify input fields";
+        default:
+            return "grpc request failed, inspect status code and service logs";
+    }
+}
+
+inline absl::string_view Explain(Subsystem, grpc::Status status) {
+    return Explain(Subsystem::kGrpc, status.error_code());
+}
+
+inline absl::string_view Explain(Subsystem, RdKafka::ErrorCode code) {
+    switch (code) {
+        case RdKafka::ERR__TRANSPORT:
+            return "broker unreachable, check connectivity and bootstrap config";
+        case RdKafka::ERR__TIMED_OUT:
+            return "operation timed out, check broker responsiveness and timeouts";
+        case RdKafka::ERR__AUTHENTICATION:
+            return "authentication failed, verify SASL/SSL credentials";
+        default:
+            return "kafka operation failed, inspect broker/client configuration";
+    }
+}
+
+inline absl::string_view Explain(Subsystem, UErrorCode code) {
+    switch (code) {
+        case U_BUFFER_OVERFLOW_ERROR:
+            return "output buffer too small for the converted string";
+        case U_INVALID_CHAR_FOUND:
+            return "invalid unicode sequence found in input data";
+        case U_MEMORY_ALLOCATION_ERROR:
+            return "memory allocation failed during unicode operation";
+        default:
+            return "ICU operation failed, verify unicode input and ICU configuration";
+    }
+}
+
+// fallback for custom codes
+
+inline absl::string_view Explain(Subsystem subsystem, absl::string_view code) {
+    if (subsystem == Subsystem::kAlgorithm) {
+        if (code == "TOKENIZE_FAIL") return "tokenization failed, input normalization may be invalid";
+        if (code == "NORMALIZE_FAIL") return "normalization failed, inspect text and ICU preprocessing";
+    }
+    return "operation failed, inspect subsystem-specific context";
+}
+
+inline absl::string_view GrpcCodeName(grpc::StatusCode code) {
+    switch (code) {
+        case grpc::StatusCode::OK: return "OK";
+        case grpc::StatusCode::CANCELLED: return "CANCELLED";
+        case grpc::StatusCode::UNKNOWN: return "UNKNOWN";
+        case grpc::StatusCode::INVALID_ARGUMENT: return "INVALID_ARGUMENT";
+        case grpc::StatusCode::DEADLINE_EXCEEDED: return "DEADLINE_EXCEEDED";
+        case grpc::StatusCode::NOT_FOUND: return "NOT_FOUND";
+        case grpc::StatusCode::ALREADY_EXISTS: return "ALREADY_EXISTS";
+        case grpc::StatusCode::PERMISSION_DENIED: return "PERMISSION_DENIED";
+        case grpc::StatusCode::RESOURCE_EXHAUSTED: return "RESOURCE_EXHAUSTED";
+        case grpc::StatusCode::FAILED_PRECONDITION: return "FAILED_PRECONDITION";
+        case grpc::StatusCode::ABORTED: return "ABORTED";
+        case grpc::StatusCode::OUT_OF_RANGE: return "OUT_OF_RANGE";
+        case grpc::StatusCode::UNIMPLEMENTED: return "UNIMPLEMENTED";
+        case grpc::StatusCode::INTERNAL: return "INTERNAL";
+        case grpc::StatusCode::UNAVAILABLE: return "UNAVAILABLE";
+        case grpc::StatusCode::DATA_LOSS: return "DATA_LOSS";
+        case grpc::StatusCode::UNAUTHENTICATED: return "UNAUTHENTICATED";
+        default: return "UNRECOGNIZED";
+    }
+}
+
+inline std::string FormatError(Subsystem subsystem, grpc::StatusCode code, absl::string_view message) {
+    const auto timespan = absl::FormatTime("%Y-%m-%dT%H:%M:%E3S%Ez", absl::Now(), GetLocalTZ());
+    return absl::StrCat(
+        "timespan=", timespan,
+        " subsystem=", SubsystemName(subsystem),
+        " code=", GrpcCodeName(code),
+        " message=\"", message, "\"",
+        " explanation=\"", Explain(subsystem, code), "\"");
+}
+
+inline std::string FormatError(Subsystem subsystem, grpc::Status status, absl::string_view message) {
+    return FormatError(subsystem, status.error_code(),
+                       absl::StrCat(message, " status_message=\"", status.error_message(), "\""));
+}
+
+inline std::string FormatError(Subsystem subsystem, RdKafka::ErrorCode code, absl::string_view message) {
+    const auto timespan = absl::FormatTime("%Y-%m-%dT%H:%M:%E3S%Ez", absl::Now(), GetLocalTZ());
+    return absl::StrCat(
+        "timespan=", timespan,
+        " subsystem=", SubsystemName(subsystem),
+        " code=", RdKafka::err2str(code),
+        " message=\"", message, "\"",
+        " explanation=\"", Explain(subsystem, code), "\"");
+}
+
+inline std::string FormatError(Subsystem subsystem, UErrorCode code, absl::string_view message) {
+    const auto timespan = absl::FormatTime("%Y-%m-%dT%H:%M:%E3S%Ez", absl::Now(), GetLocalTZ());
+    return absl::StrCat(
+        "timespan=", timespan,
+        " subsystem=", SubsystemName(subsystem),
+        " code=", u_errorName(code),
+        " message=\"", message, "\"",
+        " explanation=\"", Explain(subsystem, code), "\"");
+}
+
+inline std::string FormatError(Subsystem subsystem, absl::string_view code, absl::string_view message) {
+    const auto timespan = absl::FormatTime("%Y-%m-%dT%H:%M:%E3S%Ez", absl::Now(), GetLocalTZ());
+    return absl::StrCat(
+        "timespan=", timespan,
+        " subsystem=", SubsystemName(subsystem),
+        " code=", code,
+        " message=\"", message, "\"",
+        " explanation=\"", Explain(subsystem, code), "\"");
+}
+
+inline std::string FormatError(Subsystem subsystem, const std::string& code,
+                               absl::string_view message) {
+    return FormatError(subsystem, absl::string_view(code), message);
+}
+
+inline std::string FormatError(Subsystem subsystem, const char* code, absl::string_view message) {
+    return FormatError(subsystem, absl::string_view(code ? code : "UNKNOWN"), message);
+}
+
+}  // namespace moderation::logging
+
+#define SERVICE_LOG_ERROR(subsystem, code, message) \
+    LOG(ERROR).AtLocation(__FILE__, __LINE__) << ::moderation::logging::FormatError((subsystem), (code), (message))
+
+#define SERVICE_LOG_INFO(message) LOG(INFO).AtLocation(__FILE__, __LINE__) << (message)
+#define SERVICE_VLOG1(message) VLOG(1) << (message)
\ No newline at end of file
diff --git a/include/interceptors/logger_interceptor.hpp b/include/interceptors/logger_interceptor.hpp
new file mode 100644
index 0000000..a375bbf
--- /dev/null
+++ b/include/interceptors/logger_interceptor.hpp
@@ -0,0 +1,53 @@
+#pragma once
+#include "logger.hpp"
+
+#include <absl/strings/str_cat.h>
+#include <absl/time/clock.h>
+#include <absl/time/time.h>
+#include <grpcpp/support/interceptor.h>
+#include <grpcpp/support/server_interceptor.h>
+
+class LoggerInterceptor final : public grpc::experimental::Interceptor {
+  public:
+    explicit LoggerInterceptor(grpc::experimental::ServerRpcInfo* info)
+        : method_(info->method() ? info->method() : "unknown"),
+          start_(absl::Now()) {}
+
+    void Intercept(grpc::experimental::InterceptorBatchMethods* methods) override {
+        if (method_ == "/grpc.health.v1.Health/Check") {
+            methods->Proceed();
+            return;
+        }
+
+        const grpc::Status* status = methods->GetRecvStatus();
+        if (status != nullptr) {
+            const auto latency_ms = absl::ToInt64Milliseconds(absl::Now() - start_);
+            const auto code = status ? status->error_code() : grpc::StatusCode::UNKNOWN;
+
+            if (code == grpc::StatusCode::OK) {
+                SERVICE_VLOG1(absl::StrCat("grpc method=", method_,
+                                           " status=OK latency_ms=", latency_ms));
+            } else {
+                SERVICE_LOG_ERROR(moderation::logging::Subsystem::kGrpc, code,
+                    absl::StrCat("method=", method_,
+                                 " status_msg=", (status ? status->error_message() : "unknown"),
+                                 " latency_ms=", latency_ms));
+            }
+        }
+
+        methods->Proceed();
+    }
+
+  private:
+    std::string method_;
+    absl::Time  start_;
+};
+
+class LoggerInterceptorFactory
+    : public grpc::experimental::ServerInterceptorFactoryInterface {
+  public:
+    grpc::experimental::Interceptor*
+    CreateServerInterceptor(grpc::experimental::ServerRpcInfo* info) override {
+        return new LoggerInterceptor(info);
+    }
+};
\ No newline at end of file
diff --git a/src/moderationservice/handler/handler.cpp b/src/moderationservice/handler/handler.cpp
index a1f05a0..77b8618 100644
--- a/src/moderationservice/handler/handler.cpp
+++ b/src/moderationservice/handler/handler.cpp
@@ -1,5 +1,5 @@
 #include "handler/handler.hpp"
-#include <iostream>
+#include "interceptors/logger.hpp"
 #include <string>
 #include <utility>
 
@@ -11,8 +11,9 @@ grpc::Status ModerationHandler::ModerateObject(grpc::ServerContext* context,
                                                moderation::ModerateObjectResponse* response) {
     if (request->text().empty() || request->id() == 0) {
         response->set_success(false);
-        std::cerr << "[Handler] ERROR: Invalid moderation request. Text is empty or request ID is "
-                     "zero.\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kGrpc,
+                          grpc::StatusCode::INVALID_ARGUMENT,
+                          "invalid moderation request: text empty or request id is zero");
         return {grpc::INVALID_ARGUMENT, "Invalid moderation request"};
     }
 
@@ -23,8 +24,8 @@ grpc::Status ModerationHandler::ModerateObject(grpc::ServerContext* context,
 
     if (!success) {
         response->set_success(false);
-        std::cerr << "[Handler] ERROR: Failed to process moderation request for ID: " << requestId
-                  << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kGrpc, grpc::StatusCode::INTERNAL,
+                          absl::StrCat("failed to process moderation request request_id=", requestId));
         return {grpc::INTERNAL, "Failed to process moderation request"};
     }
 
diff --git a/src/moderationservice/kafka/kafkaclient.cpp b/src/moderationservice/kafka/kafkaclient.cpp
index 4704631..c4a1dd1 100644
--- a/src/moderationservice/kafka/kafkaclient.cpp
+++ b/src/moderationservice/kafka/kafkaclient.cpp
@@ -3,7 +3,7 @@
 #include "model/constants.hpp"
 #include "model/model_utils.hpp"
 #include "service/text_processor.hpp"
-#include <iostream>
+#include "interceptors/logger.hpp"
 #include <memory>
 #include <string>
 #include <utility>
@@ -22,16 +22,20 @@ void KafkaClient::Initialize(std::function<void(const moderation::ModerateObject
         consumer_ =
             std::make_unique<KafkaConsumer>(config_, std::move(result_callback), producer_.get());
         initialized_ = true;
-        std::cout << "KafkaClient initialized successfully.\n";
+        SERVICE_LOG_INFO("kafka producer and consumer initialized");
     } catch (const std::exception& e) {
-        std::cerr << "Failed to initialize KafkaClient: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "INITIALIZATION_FAIL", e.what());
         throw;
     }
 }
 
 bool KafkaClient::SendRequestAsync(const moderation::ModerateObjectRequest& request) {
     if (!initialized_ || !producer_) {
-        std::cerr << "KafkaClient not initialized.\n";
+        SERVICE_LOG_ERROR(
+            moderation::logging::Subsystem::kKafka,
+            "NOT_INITIALIZED",
+            "KafkaClient not initialized at requesting time"
+        );
         return false;
     }
     return producer_->SendRequestAsync(request, config_.request_topic);
@@ -39,10 +43,21 @@ bool KafkaClient::SendRequestAsync(const moderation::ModerateObjectRequest& requ
 
 void ProducerDeliveryReportCb::dr_cb(RdKafka::Message& message) {
     if (message.err() == RdKafka::ERR_NO_ERROR) {
-        std::cout << "Message delivered to topic " << message.topic_name() << ", partition "
-                  << message.partition() << ", offset " << message.offset() << "\n";
+        SERVICE_VLOG1(absl::StrCat(
+            "kafka message delivered",
+            " topic=", message.topic_name(),
+            " partition=", message.partition(),
+            " offset=", message.offset()
+        ));
     } else {
-        std::cerr << "Message delivery failed: " << message.errstr() << "\n";
+        SERVICE_LOG_ERROR(
+            moderation::logging::Subsystem::kKafka,
+            message.err(),                          
+            absl::StrCat("delivery failed"
+                        " topic=", message.topic_name(),
+                        " partition=", message.partition(),
+                        " error=", message.errstr())  
+        );
     }
 }
 
@@ -79,19 +94,22 @@ bool KafkaProducer::SendRequestAsync(const moderation::ModerateObjectRequest& re
                                      const std::string& topic) {
     std::lock_guard<std::mutex> lock(mutex_);
     if (!producer_ || topic_name_.empty()) {
-        std::cerr << "KafkaProducer not properly initialized.\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PRODUCER_NOT_INITIALIZED",
+                          "producer not properly initialized");
         return false;
     }
 
     std::string target_topic = topic.empty() ? topic_name_ : topic;
     if (target_topic.empty()) {
-        std::cerr << "KafkaProducer: No topic specified.\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "TOPIC_EMPTY",
+                          "no request topic specified");
         return false;
     }
 
     std::string serializedRequest;
     if (!request.SerializeToString(&serializedRequest)) {
-        std::cerr << "Failed to serialize ModerateObjectRequest\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "SERIALIZE_REQUEST_FAIL",
+                          "failed to serialize ModerateObjectRequest");
         return false;
     }
 
@@ -103,11 +121,13 @@ bool KafkaProducer::SendRequestAsync(const moderation::ModerateObjectRequest& re
                            key.c_str(), key.size(), 0, nullptr);
 
     if (error != RdKafka::ERR_NO_ERROR) {
-        std::cerr << "Failed to produce message: " << RdKafka::err2str(error) << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, error,
+                          absl::StrCat("failed to produce request message topic=", target_topic));
         return false;
     }
 
     producer_->poll(0);
+    SERVICE_VLOG1(absl::StrCat("kafka request produced topic=", target_topic, " key=", key));
 
     return true;
 }
@@ -117,18 +137,21 @@ bool KafkaProducer::SendResponseAsync(const moderation::ModerateObjectResponse&
     std::lock_guard<std::mutex> lock(mutex_);
 
     if (!producer_) {
-        std::cerr << "KafkaProducer not properly initialized.\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PRODUCER_NOT_INITIALIZED",
+                          "producer not properly initialized");
         return false;
     }
 
     if (topic.empty()) {
-        std::cerr << "KafkaProducer: Topic cannot be empty for SendResponseAsync\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "TOPIC_EMPTY",
+                          "response topic cannot be empty");
         return false;
     }
 
     std::string serializedResponse;
     if (!response.SerializeToString(&serializedResponse)) {
-        std::cerr << "Failed to serialize ModerateObjectResponse\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "SERIALIZE_RESPONSE_FAIL",
+                          "failed to serialize ModerateObjectResponse");
         return false;
     }
 
@@ -137,25 +160,29 @@ bool KafkaProducer::SendResponseAsync(const moderation::ModerateObjectResponse&
         serializedResponse.size(), key.c_str(), key.size(), 0, nullptr);
 
     if (error != RdKafka::ERR_NO_ERROR) {
-        std::cerr << "Failed to produce response message: " << RdKafka::err2str(error) << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, error,
+                          absl::StrCat("failed to produce response topic=", topic, " key=", key));
         return false;
     }
 
     producer_->poll(0);
+    SERVICE_VLOG1(absl::StrCat("kafka response produced topic=", topic, " key=", key));
 
     return true;
 }
 
 bool KafkaProducer::Flush(int timeoutMs) {
     if (!producer_) {
-        std::cerr << "KafkaProducer not properly initialized.\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PRODUCER_NOT_INITIALIZED",
+                          "producer not properly initialized");
         return false;
     }
 
     RdKafka::ErrorCode error = producer_->flush(timeoutMs);
 
     if (error != RdKafka::ERR_NO_ERROR) {
-        std::cerr << "Failed to flush Kafka producer: " << RdKafka::err2str(error) << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, error,
+                          "failed to flush kafka producer");
         return false;
     }
 
@@ -171,16 +198,12 @@ KafkaProducer::~KafkaProducer() {
 void ConsumerEventCb::event_cb(RdKafka::Event& event) {
     switch (event.type()) {
     case RdKafka::Event::EVENT_ERROR:
-        if (event.fatal()) {
-            std::cerr << "Fatal Kafka error: " << event.str() << "\n";
-            // Fatal errors mean consumer must be recreated
-        } else {
-            std::cerr << "Kafka error: " << event.str() << "\n";
-            // Non-fatal errors can be recovered
-        }
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, event.err(),
+                          absl::StrCat(event.fatal() ? "fatal kafka event: " : "kafka event error: ",
+                                       event.str()));
         break;
     case RdKafka::Event::EVENT_LOG:
-        std::cout << "Kafka log: " << event.str() << "\n";
+        SERVICE_VLOG1(absl::StrCat("kafka event log: ", event.str()));
         break;
     default:
         break;
@@ -198,15 +221,16 @@ void KafkaConsumer::ProcessMessage(RdKafka::Message* message) {
         // Validate that key_str contains only digits (and optional leading +/-)
         if (key_str.empty() ||
             (!std::isdigit(key_str[0]) && key_str[0] != '-' && key_str[0] != '+')) {
-            std::cerr << "Invalid message key format (not numeric): '" << key_str << "'\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "INVALID_KEY_FORMAT",
+                              absl::StrCat("message key is not numeric key=", key_str));
             return;
         }
 
         try {
             request_id = std::stoll(key_str);
         } catch (const std::exception& e) {
-            std::cerr << "Failed to parse message key '" << key_str
-                      << "' to request ID: " << e.what() << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PARSE_KEY_FAIL",
+                              absl::StrCat("failed to parse key=", key_str, " error=", e.what()));
             return;
         }
     }
@@ -216,58 +240,60 @@ void KafkaConsumer::ProcessMessage(RdKafka::Message* message) {
 
     moderation::ModerateObjectRequest request;
     if (!request.ParseFromArray(payload, static_cast<int>(payload_size))) {
-        std::cerr << "Failed to parse ModerateObjectRequest from message payload\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PARSE_REQUEST_FAIL",
+                          "failed to parse ModerateObjectRequest from payload");
         return;
     }
 
     moderation::ObjectType object_type = request.type();
 
     if (!moderation::ObjectType_IsValid(static_cast<int>(object_type))) {
-        std::cerr << "Received ModerateObjectRequest with invalid ObjectType for request ID "
-                  << request_id << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "INVALID_OBJECT_TYPE",
+                          absl::StrCat("invalid object type request_id=", request_id));
         return;
     }
 
     if (object_type == moderation::OBJECT_TYPE_UNSPECIFIED) {
-        std::cerr << "Received ModerateObjectRequest with unspecified ObjectType for request ID "
-                  << request_id << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "UNSPECIFIED_OBJECT_TYPE",
+                          absl::StrCat("unspecified object type request_id=", request_id));
         return;
     }
 
     int type_value = static_cast<int>(object_type);
 
-    std::cout << "Received ModerateObjectRequest for request ID " << request_id
-              << ", ObjectType: " << type_value << " ("
-              << moderation::utils::ObjectTypeToString(object_type) << ")"
-              << ", text: " << request.text() << "\n";
+    SERVICE_VLOG1(absl::StrCat("kafka request consumed request_id=", request_id,
+                               " object_type=", type_value, " (",
+                               moderation::utils::ObjectTypeToString(object_type), ")"));
 
     bool isFlagged = false;
 
     try {
         isFlagged = TextProcessor::TextProcessing(request.text());
     } catch (const std::exception& e) {
-        std::cerr << "Error processing text: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kAlgorithm, "TEXT_PROCESSING_FAIL",
+                          e.what());
         return;
     }
 
     moderation::ModerateObjectResponse response;
     response.set_success(isFlagged);
 
-    std::cout << "Text processing result for request ID " << request_id << ": "
-              << (isFlagged ? "FLAGGED" : "PASSED") << "\n";
+    SERVICE_VLOG1(absl::StrCat("text processing request_id=", request_id,
+                               " result=", (isFlagged ? "FLAGGED" : "PASSED")));
 
     if (producer_ != nullptr) {
         std::string response_key = std::to_string(request_id);
         if (!producer_->SendResponseAsync(response, config_.result_topic, response_key)) {
-            std::cerr << "Failed to send moderation response to Kafka for request ID: "
-                      << request_id << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "SEND_RESPONSE_FAIL",
+                              absl::StrCat("failed to send moderation response request_id=",
+                                           request_id));
         } else {
-            std::cout << "Sent moderation response to topic " << config_.result_topic
-                      << " for request ID: " << request_id << "\n";
+            SERVICE_VLOG1(absl::StrCat("sent moderation response topic=", config_.result_topic,
+                                       " request_id=", request_id));
         }
     } else {
-        std::cerr << "Producer not available, cannot send response for request ID: " << request_id
-                  << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "PRODUCER_NOT_AVAILABLE",
+                          absl::StrCat("producer unavailable request_id=", request_id));
     }
 
     if (callback_) {
@@ -309,7 +335,7 @@ KafkaConsumer::KafkaConsumer(const KafkaConfig& config, MessageCallback callback
         throw std::runtime_error("Failed to subscribe to Kafka topics: " + RdKafka::err2str(error));
     }
 
-    std::cout << "Kafka consumer subscribed to topic: " << config.request_topic << "\n";
+    SERVICE_LOG_INFO(absl::StrCat("kafka consumer subscribed topic=", config.request_topic));
 
     running_ = false;
 }
@@ -335,15 +361,17 @@ void KafkaConsumer::ConsumeLoop() {
         if (message->err() == RdKafka::ERR_NO_ERROR) {
             ProcessMessage(message);
         } else if (message->err() == RdKafka::ERR__PARTITION_EOF) {
-            std::cout << "Reached end of partition for topic " << message->topic_name() << "\n";
+            SERVICE_VLOG1(
+                absl::StrCat("kafka partition end of topic=", message->topic_name()));
         } else if (message->err() == RdKafka::ERR__TIMED_OUT) {
         } else {
-            std::cerr << "Kafka consumer error: " << message->errstr() << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, message->err(),
+                              absl::StrCat("kafka consumer error: ", message->errstr()));
         }
 
         delete message;
     }
-    std::cout << "Kafka consumer ended." << "\n";
+    SERVICE_LOG_INFO("kafka consumer stopped");
 }
 
 KafkaConsumer::~KafkaConsumer() {
@@ -366,7 +394,8 @@ void KafkaConsumer::Stop() {
 
 void KafkaClient::StartConsumer() {
     if (!initialized_ || !consumer_) {
-        std::cerr << "KafkaClient not initialized." << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "NOT_INITIALIZED",
+                          "kafka client not initialized");
         return;
     }
     consumer_->Start();
@@ -374,7 +403,8 @@ void KafkaClient::StartConsumer() {
 
 void KafkaClient::StopConsumer() {
     if (!initialized_ || !consumer_) {
-        std::cerr << "KafkaClient not initialized." << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "NOT_INITIALIZED",
+                          "kafka client not initialized");
         return;
     }
     consumer_->Stop();
@@ -391,12 +421,13 @@ void KafkaClient::Shutdown() {
     producer_.reset();
 
     initialized_ = false;
-    std::cout << "KafkaClient shutdown completed." << "\n";
+    SERVICE_LOG_INFO("kafka client shutdown completed");
 }
 
 void KafkaClient::Flush() {
     if (!initialized_ || !producer_) {
-        std::cerr << "KafkaClient not initialized." << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "NOT_INITIALIZED",
+                          "kafka client not initialized");
         return;
     }
     producer_->Flush(moderation::config::KAFKA_FLUSH_TIMEOUT_MS);
diff --git a/src/moderationservice/main.cpp b/src/moderationservice/main.cpp
index 90da759..d1accdb 100644
--- a/src/moderationservice/main.cpp
+++ b/src/moderationservice/main.cpp
@@ -1,17 +1,18 @@
 #include "config/config.hpp"
 #include "handler/handler.hpp"
+#include "interceptors/logger.hpp"
 #include "repository/repository.hpp"
 #include "server/server.hpp"
 #include "service/service.hpp"
+#include "interceptors/log_sink.hpp"
 #include <csignal>
-#include <iostream>
 #include <memory>
 
 std::atomic<bool> shutdown_requested(false);
 std::shared_ptr<Server> global_server_ptr;
 
 void signalHandler(int signum) {
-    std::cout << "Interrupt signal (" << signum << ") received.\n";
+    SERVICE_LOG_INFO(absl::StrCat("interrupt signal received signal=", signum));
     shutdown_requested.store(true);
 
     if (global_server_ptr) {
@@ -20,6 +21,9 @@ void signalHandler(int signum) {
 }
 
 int main() {
+    moderation::logging::InstallReadableSink();
+    grpc_init();
+    moderation::logging::ActivateReadableSink();
     try {
         std::signal(SIGINT, signalHandler);  // Handle Ctrl+C
         std::signal(SIGTERM, signalHandler); // Docker/Kubernetes termination signal
@@ -49,17 +53,19 @@ int main() {
             std::this_thread::sleep_for(std::chrono::seconds(1));
         }
 
-        std::cout << "Shutting down server...\n";
+        SERVICE_LOG_INFO("shutting down server");
 
         kafkaClient->StopConsumer();
         kafkaClient->Flush();
         kafkaClient->Shutdown();
 
-        std::cout << "Server shutdown complete.\n";
+        SERVICE_LOG_INFO("server shutdown complete");
     } catch (const std::exception& e) {
-        std::cerr << "Main Fatal Error: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kServer, "MAIN_FATAL_ERROR", e.what());
+        grpc_shutdown();
         return 1;
     }
+    grpc_shutdown();
 
     return 0;
 }
diff --git a/src/moderationservice/repository/repository.cpp b/src/moderationservice/repository/repository.cpp
index a928c0d..5fb43a7 100644
--- a/src/moderationservice/repository/repository.cpp
+++ b/src/moderationservice/repository/repository.cpp
@@ -1,5 +1,5 @@
 #include "repository/repository.hpp"
-#include <iostream>
+#include "interceptors/logger.hpp"
 #include <memory>
 #include <string>
 
@@ -9,7 +9,8 @@ ModerationRepository::ModerationRepository(const std::string& database_url)
     try {
         db_connection_ = std::make_unique<pqxx::connection>(database_url_);
     } catch (const std::exception& e) {
-        std::cerr << "Database connection error: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kRepository, "DB_CONNECTION_FAIL",
+                          e.what());
         throw;
     }
 }
@@ -19,7 +20,8 @@ ModerationRepository::~ModerationRepository() = default;
 bool ModerationRepository::SaveModerationResult(const ModerationRecord& result) {
     try {
         if (!db_connection_ || !db_connection_->is_open()) {
-            std::cerr << "Database connection is not open." << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kRepository, "DB_NOT_OPEN",
+                              "database connection is not open");
             return false;
         }
 
@@ -45,7 +47,7 @@ bool ModerationRepository::SaveModerationResult(const ModerationRecord& result)
         transaction.commit();
         return true;
     } catch (const std::exception& e) {
-        std::cerr << "Error saving moderation result: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kRepository, "DB_SAVE_FAIL", e.what());
         return false;
     }
 }
diff --git a/src/moderationservice/server/server.cpp b/src/moderationservice/server/server.cpp
index c684059..ba8bb04 100644
--- a/src/moderationservice/server/server.cpp
+++ b/src/moderationservice/server/server.cpp
@@ -1,6 +1,6 @@
 #include "server/server.hpp"
+#include "interceptors/logger.hpp"
 #include <grpcpp/ext/proto_server_reflection_plugin.h>
-#include <iostream>
 #include <memory>
 #include <string>
 #include <utility>
@@ -24,12 +24,13 @@ void Server::Start() {
     builder.experimental().SetInterceptorCreators(std::move(this->interceptor_creators_));
 
     this->server_ = builder.BuildAndStart();
-    std::cout << this->server_name_ << " listening on " << this->server_address_ << "\n";
+    SERVICE_LOG_INFO(
+        absl::StrCat(this->server_name_, " listening on ", this->server_address_));
     this->server_->Wait();
 }
 
 void Server::Stop() {
-    std::cout << "Shutting down " << this->server_name_ << "..." << "\n";
+    SERVICE_LOG_INFO(absl::StrCat("shutting down ", this->server_name_));
     this->server_->Shutdown();
-    std::cout << this->server_name_ << " shut down." << "\n";
+    SERVICE_LOG_INFO(absl::StrCat(this->server_name_, " shut down"));
 }
diff --git a/src/moderationservice/server/server.hpp b/src/moderationservice/server/server.hpp
index 3b4b0b0..2d582db 100644
--- a/src/moderationservice/server/server.hpp
+++ b/src/moderationservice/server/server.hpp
@@ -1,6 +1,6 @@
 #pragma once
 
-#include "interceptors/logger.hpp"
+#include "interceptors/logger_interceptor.hpp"
 #include <grpcpp/grpcpp.h>
 #include <grpcpp/impl/service_type.h>
 #include <grpcpp/security/server_credentials.h>
diff --git a/src/moderationservice/service/service.cpp b/src/moderationservice/service/service.cpp
index 146e525..85ec0ec 100644
--- a/src/moderationservice/service/service.cpp
+++ b/src/moderationservice/service/service.cpp
@@ -1,6 +1,6 @@
 #include "service/service.hpp"
+#include "interceptors/logger.hpp"
 #include "model/model_utils.hpp"
-#include <iostream>
 #include <memory>
 #include <string>
 #include <utility>
@@ -18,10 +18,12 @@ void ModerationService::InitializeKafkaCallback() {
             if (auto self = weakThis.lock()) {
                 self->HandleModerationResult(response, request_id, originalText, object_type);
             } else {
-                std::cerr << "ModerationService expired, cannot handle result" << "\n";
+                SERVICE_LOG_ERROR(moderation::logging::Subsystem::kService, "SERVICE_EXPIRED",
+                                  "ModerationService expired, cannot handle result");
             }
         } catch (const std::exception& e) {
-            std::cerr << "Error in Kafka callback: " << e.what() << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "KAFKA_CALLBACK_FAIL",
+                              e.what());
         }
     };
     kafkaClient_->Initialize(callback);
@@ -29,9 +31,8 @@ void ModerationService::InitializeKafkaCallback() {
 bool ModerationService::ProcessModerationRequest(int64_t request_id, const std::string& text) {
     try {
         if (text.empty() || request_id == 0) {
-            std::cerr << "[Service] ERROR: Invalid moderation request. Text is empty or request ID "
-                         "is zero."
-                      << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kService, grpc::StatusCode::INVALID_ARGUMENT,
+                              "invalid moderation request: text empty or request_id is zero");
             return false;
         }
 
@@ -40,14 +41,15 @@ bool ModerationService::ProcessModerationRequest(int64_t request_id, const std::
         kafka_request.set_text(text);
 
         if (!kafkaClient_->SendRequestAsync(kafka_request)) {
-            std::cerr << "[Service] ERROR: Failed to send moderation request to Kafka."
-                      << request_id << "\n";
+            SERVICE_LOG_ERROR(moderation::logging::Subsystem::kKafka, "SEND_REQUEST_FAIL",
+                              absl::StrCat("failed to send moderation request request_id=", request_id));
             return false;
         }
 
         return true;
     } catch (const std::exception& e) {
-        std::cerr << "[Service] ERROR: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kService, grpc::StatusCode::INTERNAL,
+                          e.what());
         return false;
     }
 }
@@ -61,14 +63,14 @@ void ModerationService::HandleModerationResult(const moderation::ModerateObjectR
             std::string reason =
                 "Violates content policy"; // This can be enhanced to provide specific
                                            // reasons based on the response
-            std::cout << "Text flagged for moderation, request id: " << request_id << "\n";
+            SERVICE_VLOG1(absl::StrCat("text flagged request_id=", request_id));
             SaveResultToDatabase(request_id, originalText, isFlagged, reason, objectType);
 
         } else {
-            std::cout << "Text passed moderation: " << request_id << "\n";
+            SERVICE_VLOG1(absl::StrCat("text passed request_id=", request_id));
         }
     } catch (const std::exception& e) {
-        std::cerr << "Error handling moderation result: " << e.what() << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kService, "HANDLE_RESULT_FAIL", e.what());
     }
 }
 
@@ -84,12 +86,11 @@ void ModerationService::SaveResultToDatabase(int64_t object_id, const std::strin
     record.reason = reason;
 
     if (!repository_->SaveModerationResult(record)) {
-        std::cerr << "Failed to save moderation result to database for object ID: " << object_id
-                  << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kRepository, "SAVE_RESULT_FAIL",
+                          absl::StrCat("failed to save moderation result object_id=", object_id));
     } else {
         std::string type_name = moderation::utils::ObjectTypeToString(object_type);
-
-        std::cout << "Moderation result saved to database for object ID: " << object_id
-                  << ", Type: " << type_name << "\n";
+        SERVICE_VLOG1(absl::StrCat("moderation result saved object_id=", object_id,
+                                   " type=", type_name));
     }
 }
diff --git a/src/moderationservice/service/text_normalization.cpp b/src/moderationservice/service/text_normalization.cpp
index 5a068ae..c4c588b 100644
--- a/src/moderationservice/service/text_normalization.cpp
+++ b/src/moderationservice/service/text_normalization.cpp
@@ -1,6 +1,6 @@
 #include "service/text_normalization.hpp"
+#include "interceptors/logger.hpp"
 #include "model/unicode_constants.hpp"
-#include <iostream>
 #include <string>
 #include <vector>
 
@@ -13,14 +13,16 @@ std::string TextNormalization(const std::string& textN) {
     const icu::Normalizer2* normalizer = icu::Normalizer2::getNFCInstance(status);
 
     if (U_FAILURE(status) != 0) {
-        std::cerr << "Error getting normalizer" << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kIcu, status,
+                          "failed to get ICU NFC normalizer");
         return "";
     }
 
     icu::UnicodeString normalized = normalizer->normalize(text, status);
 
     if (U_FAILURE(status) != 0) {
-        std::cerr << "Error normalizing" << "\n";
+        SERVICE_LOG_ERROR(moderation::logging::Subsystem::kIcu, status,
+                          "failed to normalize text with ICU");
         return "";
     }