Introduce debugs logs for ease the troubleshooting efforts

components/websubhub-consolidator/Dependencies.toml

@@ -324,7 +324,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "websubhub"
-version = "1.15.0"
+version = "1.15.1"
 dependencies = [
 	{org = "ballerina", name = "crypto"},
 	{org = "ballerina", name = "http"},

components/websubhub-consolidator/consolidator_service.bal

@@ -30,7 +30,7 @@ http:Service consolidatorService = service object {
             topics: getTopics(),
             subscriptions: getSubscriptions()
         };
-        log:printInfo("Request received to retrieve state-snapshot, hence responding with the current state-snapshot", state = stateSnapshot);
+        log:printDebug("Request received to retrieve state-snapshot, hence responding with the current state-snapshot", state = stateSnapshot);
         return stateSnapshot;
     }
 };
@@ -39,46 +39,74 @@ isolated function consolidateSystemState() returns error? {
     do {
         while true {
             kafka:BytesConsumerRecord[] records = check conn:websubEventConsumer->poll(config:kafka.consumer.pollingInterval);
+            log:printDebug("Polled Kafka records for state consolidation", recordCount = records.length(), pollingInterval = config:kafka.consumer.pollingInterval);
+
+            if records.length() > 0 {
+                log:printDebug("Processing batch of Kafka records", batchSize = records.length());
+            }
+
             foreach kafka:BytesConsumerRecord currentRecord in records {
                 string lastPersistedData = check string:fromBytes(currentRecord.value);
+                int messageSize = lastPersistedData.length();
+                log:printDebug("Processing Kafka record for consolidation", messageSize = messageSize, offset = currentRecord.offset);
                 error? result = processPersistedData(lastPersistedData);
                 if result is error {
-                    log:printError("Error occurred while processing received event ", 'error = result);
+                    common:logError("Error occurred while processing received event ", result);
+                } else {
+                    log:printDebug("Successfully processed Kafka record", offset = currentRecord.offset);
                 }
             }
         }
     } on fail var e {
+        log:printDebug("Error in consolidation loop, closing Kafka consumer", gracePeriod = config:kafka.consumer.gracefulClosePeriod);
         _ = check conn:websubEventConsumer->close(config:kafka.consumer.gracefulClosePeriod);
         return e;
     }
 }
 
 isolated function processPersistedData(string persistedData) returns error? {
+    log:printDebug("Starting persisted data processing", dataSize = persistedData.length());
+
     json payload = check value:fromJsonString(persistedData);
     string hubMode = check payload.hubMode;
+    log:printDebug("Processing event based on hub mode", hubMode = hubMode);
+
     match hubMode {
         "register" => {
+            log:printDebug("Processing topic registration event");
             check processTopicRegistration(payload);
+            log:printDebug("Topic registration event processed successfully");
         }
         "deregister" => {
+            log:printDebug("Processing topic deregistration event");
             check processTopicDeregistration(payload);
+            log:printDebug("Topic deregistration event processed successfully");
         }
         "subscribe" => {
+            log:printDebug("Processing subscription event");
             check processSubscription(payload);
+            log:printDebug("Subscription event processed successfully");
         }
         "unsubscribe" => {
+            log:printDebug("Processing unsubscription event");
             check processUnsubscription(payload);
+            log:printDebug("Unsubscription event processed successfully");
         }
         _ => {
+            common:logError("Invalid hub mode received", hubMode = hubMode);
             return error(string `Error occurred while deserializing subscriber events with invalid hubMode [${hubMode}]`);
         }
     }
+    log:printDebug("Completed processing persisted data", hubMode = hubMode);
 }
 
 isolated function processStateUpdate() returns error? {
+    log:printDebug("Processing state update - gathering current system state");
     common:SystemStateSnapshot stateSnapshot = {
         topics: getTopics(),
         subscriptions: getSubscriptions()
     };
+    log:printDebug("Created state snapshot for persistence", topicCount = stateSnapshot.topics.length(), subscriptionCount = stateSnapshot.subscriptions.length());
     check persist:persistWebsubEventsSnapshot(stateSnapshot);
+    log:printDebug("State snapshot persisted successfully");
 }

components/websubhub-consolidator/init_consolidator.bal

@@ -27,7 +27,7 @@ public function main() returns error? {
     // Initialize consolidator-service state
     error? stateSyncResult = syncSystemState();
     if stateSyncResult is error {
-        common:logError("Error while syncing system state during startup", stateSyncResult, "FATAL");
+        common:logError("Error while syncing system state during startup", stateSyncResult, severity = "FATAL");
         return;
     }
 
@@ -67,7 +67,7 @@ isolated function syncSystemState() returns error? {
             check persist:persistWebsubEventsSnapshot(lastStateSnapshot);
         }
     } on fail error kafkaError {
-        common:logError("Error occurred while syncing system-state", kafkaError, "FATAL");
+        common:logError("Error occurred while syncing system-state", kafkaError, severity = "FATAL");
         error? result = check websubEventsSnapshotConsumer->close(config:kafka.consumer.gracefulClosePeriod);
         if result is error {
             common:logError("Error occurred while gracefully closing kafka:Consumer", result);

components/websubhub-consolidator/modules/common/utils.bal

@@ -13,6 +13,7 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
+
 import ballerina/log;
 import ballerina/random;
 
@@ -51,15 +52,12 @@ public isolated function generateRandomString() returns string {
 
 # Logs errors with proper details.
 #
-# + baseErrorMsg - Base error message  
-# + err - Current error
-# + severity - Severity of the error
-public isolated function logError(string baseErrorMsg, error err, string severity = "RECOVERABLE") {
-    string errorMsg = string `${baseErrorMsg}: ${err.message()}`;
-    error? cause = err.cause();
-    while cause is error {
-        errorMsg += string `: ${cause.message()}`;
-        cause = cause.cause();
+# + msg - Base error message  
+# + error - Current error
+# + keyValues - Additional key values to be logged
+public isolated function logError(string msg, error? 'error = (), *log:KeyValues keyValues) {
+    if !keyValues.hasKey("severity") {
+        keyValues["severity"] = "RECOVERABLE";
     }
-    log:printError(errorMsg, severity = severity, stackTrace = err.stackTrace());
+    log:printError(msg, 'error, keyValues = keyValues);
 }

components/websubhub-consolidator/modules/persistence/persistence.bal

@@ -18,13 +18,19 @@ import websubhub.consolidator.common;
 import websubhub.consolidator.config;
 import websubhub.consolidator.connections as conn;
 
+import ballerina/log;
+
 public isolated function persistWebsubEventsSnapshot(common:SystemStateSnapshot systemStateSnapshot) returns error? {
+    log:printDebug("Persisting system state snapshot", topicCount = systemStateSnapshot.topics.length(), subscriptionCount = systemStateSnapshot.subscriptions.length());
     json payload = systemStateSnapshot.toJson();
     check produceKafkaMessage(config:state.snapshot.topic, payload);
+    log:printDebug("System state snapshot persisted successfully", targetTopic = config:state.snapshot.topic);
 }
 
 isolated function produceKafkaMessage(string topicName, json payload) returns error? {
+    log:printDebug("Producing Kafka message", targetTopic = topicName);
     byte[] serializedContent = payload.toJsonString().toBytes();
     check conn:statePersistProducer->send({topic: topicName, value: serializedContent});
     check conn:statePersistProducer->'flush();
+    log:printDebug("Message successfully sent to Kafka", topic = topicName);
 }

components/websubhub-consolidator/websub_subscribers.bal

@@ -16,54 +16,63 @@
 
 import websubhub.consolidator.common;
 
-import ballerina/lang.value;
+import ballerina/log;
 import ballerina/websubhub;
 
 isolated map<websubhub:VerifiedSubscription> subscribersCache = {};
 
-isolated function deSerializeSubscribersMessage(string lastPersistedData) returns websubhub:VerifiedSubscription[]|error {
-    websubhub:VerifiedSubscription[] currentSubscriptions = [];
-    json[] payload = <json[]>check value:fromJsonString(lastPersistedData);
-    foreach var data in payload {
-        websubhub:VerifiedSubscription subscription = check data.cloneWithType(websubhub:VerifiedSubscription);
-        currentSubscriptions.push(subscription);
-    }
-    return currentSubscriptions;
-}
-
 isolated function refreshSubscribersCache(websubhub:VerifiedSubscription[] persistedSubscribers) {
+    log:printDebug("Refreshing subscribers cache from persisted data", persistedSubscriberCount = persistedSubscribers.length());
     foreach var subscriber in persistedSubscribers {
         string groupName = common:generatedSubscriberId(subscriber.hubTopic, subscriber.hubCallback);
         lock {
             subscribersCache[groupName] = subscriber.cloneReadOnly();
         }
+        log:printDebug("Added subscriber to cache during refresh", subscriberId = groupName, topic = subscriber.hubTopic, callback = subscriber.hubCallback);
     }
+    log:printDebug("Subscribers cache refresh completed", totalCachedSubscribers = subscribersCache.length());
 }
 
 isolated function processSubscription(json payload) returns error? {
+    log:printDebug("Processing subscription event");
     websubhub:VerifiedSubscription subscription = check payload.cloneWithType(websubhub:VerifiedSubscription);
     string subscriberId = common:generatedSubscriberId(subscription.hubTopic, subscription.hubCallback);
+    log:printDebug("Deserialized subscription", subscriberId = subscriberId, topic = subscription.hubTopic, callback = subscription.hubCallback);
+    boolean subscriptionAdded = false;
     lock {
         // add the subscriber if subscription event received
         if !subscribersCache.hasKey(subscriberId) {
             subscribersCache[subscriberId] = subscription.cloneReadOnly();
+            subscriptionAdded = true;
+            log:printDebug("Added new subscription to cache", subscriberId = subscriberId, totalSubscriptions = subscribersCache.length());
+        } else {
+            log:printDebug("Subscription already exists in cache, skipping", subscriberId = subscriberId);
         }
     }
     check processStateUpdate();
+    log:printDebug("Subscription processing completed", subscriberId = subscriberId, wasAdded = subscriptionAdded);
 }
 
 isolated function processUnsubscription(json payload) returns error? {
+    log:printDebug("Processing unsubscription event");
     websubhub:VerifiedUnsubscription unsubscription = check payload.cloneWithType(websubhub:VerifiedUnsubscription);
     string subscriberId = common:generatedSubscriberId(unsubscription.hubTopic, unsubscription.hubCallback);
+    log:printDebug("Deserialized unsubscription", subscriberId = subscriberId, topic = unsubscription.hubTopic, callback = unsubscription.hubCallback);
     lock {
         // remove the subscriber if the unsubscription event received
-        _ = subscribersCache.removeIfHasKey(subscriberId);
+        websubhub:VerifiedSubscription? removedSubscription = subscribersCache.removeIfHasKey(subscriberId);
+        boolean subscriptionRemoved = removedSubscription is websubhub:VerifiedSubscription;
+        log:printDebug("Removed subscription from cache", subscriberId = subscriberId, wasRemoved = subscriptionRemoved, totalSubscriptions = subscribersCache.length());
     }
     check processStateUpdate();
+    log:printDebug("Unsubscription processing completed", subscriberId = subscriberId);
 }
 
 isolated function getSubscriptions() returns websubhub:VerifiedSubscription[] {
+    websubhub:VerifiedSubscription[] subscriptions;
     lock {
-        return subscribersCache.toArray().cloneReadOnly();
+        subscriptions = subscribersCache.toArray().cloneReadOnly();
     }
+    log:printDebug("Retrieved subscriptions from cache", subscriptionCount = subscriptions.length());
+    return subscriptions;
 }

components/websubhub-consolidator/websub_topics.bal

@@ -15,38 +15,55 @@
 // under the License.
 
 import ballerina/lang.value;
+import ballerina/log;
 import ballerina/websubhub;
 
 isolated map<websubhub:TopicRegistration> registeredTopicsCache = {};
 
 isolated function refreshTopicCache(websubhub:TopicRegistration[] persistedTopics) {
+    log:printDebug("Refreshing topic cache from persisted data", persistedTopicCount = persistedTopics.length());
     foreach var topic in persistedTopics.cloneReadOnly() {
         lock {
             registeredTopicsCache[topic.topic] = topic.cloneReadOnly();
         }
+        log:printDebug("Added topic to cache during refresh", topicName = topic.topic);
     }
+    log:printDebug("Topic cache refresh completed", totalCachedTopics = registeredTopicsCache.length());
 }
 
 isolated function processTopicRegistration(json payload) returns error? {
+    log:printDebug("Processing topic registration event");
     websubhub:TopicRegistration registration = check value:cloneWithType(payload);
+    log:printDebug("Deserialized topic registration", topicName = registration.topic);
     lock {
         // add the topic if topic-registration event received
         registeredTopicsCache[registration.topic] = registration.cloneReadOnly();
     }
+    log:printDebug("Added topic to cache", topicName = registration.topic, totalTopics = registeredTopicsCache.length());
     check processStateUpdate();
+    log:printDebug("Topic registration processing completed", topicName = registration.topic);
 }
 
 isolated function processTopicDeregistration(json payload) returns error? {
+    log:printDebug("Processing topic deregistration event");
     websubhub:TopicDeregistration deregistration = check value:cloneWithType(payload);
+    log:printDebug("Deserialized topic deregistration", topicName = deregistration.topic);
+    boolean topicRemoved = false;
     lock {
         // remove the topic if topic-deregistration event received
-        _ = registeredTopicsCache.removeIfHasKey(deregistration.topic);
+        websubhub:TopicRegistration? removedTopic = registeredTopicsCache.removeIfHasKey(deregistration.topic);
+        topicRemoved = removedTopic is websubhub:TopicRegistration;
     }
+    log:printDebug("Removed topic from cache", topicName = deregistration.topic, wasRemoved = topicRemoved, totalTopics = registeredTopicsCache.length());
     check processStateUpdate();
+    log:printDebug("Topic deregistration processing completed", topicName = deregistration.topic);
 }
 
 isolated function getTopics() returns websubhub:TopicRegistration[] {
+    websubhub:TopicRegistration[] topics;
     lock {
-        return registeredTopicsCache.toArray().cloneReadOnly();
+        topics = registeredTopicsCache.toArray().cloneReadOnly();
     }
+    log:printDebug("Retrieved topics from cache", topicCount = topics.length());
+    return topics;
 }

components/websubhub/Dependencies.toml

@@ -333,7 +333,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "websubhub"
-version = "1.15.0"
+version = "1.15.1"
 dependencies = [
 	{org = "ballerina", name = "crypto"},
 	{org = "ballerina", name = "http"},