Refactored logging for consistency (#524)

* Remove stack traces and exception causes from logs

Signed-off-by: Daniel Widdis <[email protected]>

* Logs no longer reveal transport action names

Signed-off-by: Daniel Widdis <[email protected]>

* Rebase, remove unused param and logger, typo fix

Signed-off-by: Daniel Widdis <[email protected]>

* Add logging for all client Transport calls

Signed-off-by: Daniel Widdis <[email protected]>

* Add back stack traces in error logging for failed transport queries

Signed-off-by: Daniel Widdis <[email protected]>

* Add back in stack traces for other error messages outside transport

Signed-off-by: Daniel Widdis <[email protected]>

* Add id and other details to workflow step error messages

Signed-off-by: Daniel Widdis <[email protected]>

---------

Signed-off-by: Daniel Widdis <[email protected]>

Author: dbwiddis

CHANGELOG.md

@@ -20,9 +20,11 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 - Substitute REST path or body parameters in Workflow Steps ([#525](https://github.com/opensearch-project/flow-framework/pull/525))
 - Added an optional workflow_step param to the get workflow steps API ([#538](https://github.com/opensearch-project/flow-framework/pull/538))
 
+### Enhancements
 ### Bug Fixes
 ### Infrastructure
 ### Documentation
 ### Maintenance
 ### Refactoring
 - Moved workflow-steps.json to Enum ([#523](https://github.com/opensearch-project/flow-framework/pull/523))
+- Refactored logging for consistency ([#524](https://github.com/opensearch-project/flow-framework/pull/524))

src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java

@@ -180,19 +180,20 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
                 @SuppressWarnings("deprecation")
                 ActionListener<CreateIndexResponse> actionListener = ActionListener.wrap(r -> {
                     if (r.isAcknowledged()) {
-                        logger.info("create index:{}", indexName);
+                        logger.info("create index: {}", indexName);
                         internalListener.onResponse(true);
                     } else {
                         internalListener.onResponse(false);
                     }
                 }, e -> {
-                    logger.error("Failed to create index {}", indexName, e);
-                    internalListener.onFailure(new FlowFrameworkException(e.getMessage(), ExceptionsHelper.status(e)));
+                    String errorMessage = "Failed to create index " + indexName;
+                    logger.error(errorMessage, e);
+                    internalListener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
                 });
                 CreateIndexRequest request = new CreateIndexRequest(indexName).mapping(mapping).settings(indexSettings);
                 client.admin().indices().create(request, actionListener);
             } else {
-                logger.debug("index:{} is already created", indexName);
+                logger.debug("index: {} is already created", indexName);
                 if (indexMappingUpdated.containsKey(indexName) && !indexMappingUpdated.get(indexName).get()) {
                     shouldUpdateIndex(indexName, index.getVersion(), ActionListener.wrap(r -> {
                         if (r) {
@@ -223,10 +224,7 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
                                                     String errorMessage = "Failed to update index setting for: " + indexName;
                                                     logger.error(errorMessage, exception);
                                                     internalListener.onFailure(
-                                                        new FlowFrameworkException(
-                                                            errorMessage + " : " + exception.getMessage(),
-                                                            ExceptionsHelper.status(exception)
-                                                        )
+                                                        new FlowFrameworkException(errorMessage, ExceptionsHelper.status(exception))
                                                     );
                                                 }));
                                         } else {
@@ -238,10 +236,7 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
                                         String errorMessage = "Failed to update index " + indexName;
                                         logger.error(errorMessage, exception);
                                         internalListener.onFailure(
-                                            new FlowFrameworkException(
-                                                errorMessage + " : " + exception.getMessage(),
-                                                ExceptionsHelper.status(exception)
-                                            )
+                                            new FlowFrameworkException(errorMessage, ExceptionsHelper.status(exception))
                                         );
                                     })
                                 );
@@ -251,11 +246,9 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
                             internalListener.onResponse(true);
                         }
                     }, e -> {
-                        String errorMessage = "Failed to update index mapping";
+                        String errorMessage = "Failed to update index mapping for " + indexName;
                         logger.error(errorMessage, e);
-                        internalListener.onFailure(
-                            new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e))
-                        );
+                        internalListener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
                     }));
                 } else {
                     // No need to update index if it's already updated.
@@ -265,7 +258,7 @@ public void initFlowFrameworkIndexIfAbsent(FlowFrameworkIndex index, ActionListe
         } catch (Exception e) {
             String errorMessage = "Failed to init index " + indexName;
             logger.error(errorMessage, e);
-            listener.onFailure(new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e)));
+            listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
         }
     }
 
@@ -328,11 +321,11 @@ public void putTemplateToGlobalContext(Template template, ActionListener<IndexRe
                 client.index(request, ActionListener.runBefore(listener, context::restore));
             } catch (Exception e) {
                 String errorMessage = "Failed to index global_context index";
-                logger.error(errorMessage);
-                listener.onFailure(new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e)));
+                logger.error(errorMessage, e);
+                listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
             }
         }, e -> {
-            logger.error("Failed to create global_context index", e);
+            logger.error("Failed to create global_context index");
             listener.onFailure(e);
         }));
     }
@@ -349,7 +342,7 @@ public void initializeConfigIndex(ActionListener<Boolean> listener) {
             }
             encryptorUtils.initializeMasterKey(listener);
         }, createIndexException -> {
-            logger.error("Failed to create config index", createIndexException);
+            logger.error("Failed to create config index");
             listener.onFailure(createIndexException);
         }));
     }
@@ -385,13 +378,13 @@ public void putInitialStateToWorkflowState(String workflowId, User user, ActionL
             } catch (Exception e) {
                 String errorMessage = "Failed to put state index document";
                 logger.error(errorMessage, e);
-                listener.onFailure(new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e)));
+                listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
             }
 
         }, e -> {
             String errorMessage = "Failed to create workflow_state index";
             logger.error(errorMessage, e);
-            listener.onFailure(new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e)));
+            listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
         }));
     }
 
@@ -403,11 +396,9 @@ public void putInitialStateToWorkflowState(String workflowId, User user, ActionL
      */
     public void updateTemplateInGlobalContext(String documentId, Template template, ActionListener<IndexResponse> listener) {
         if (!doesIndexExist(GLOBAL_CONTEXT_INDEX)) {
-            String exceptionMessage = "Failed to update template for workflow_id : "
-                + documentId
-                + ", global_context index does not exist.";
-            logger.error(exceptionMessage);
-            listener.onFailure(new FlowFrameworkException(exceptionMessage, RestStatus.BAD_REQUEST));
+            String errorMessage = "Failed to update template for workflow_id : " + documentId + ", global_context index does not exist.";
+            logger.error(errorMessage);
+            listener.onFailure(new FlowFrameworkException(errorMessage, RestStatus.BAD_REQUEST));
             return;
         }
         doesTemplateExist(documentId, templateExists -> {
@@ -426,9 +417,7 @@ public void updateTemplateInGlobalContext(String documentId, Template template,
                         } catch (Exception e) {
                             String errorMessage = "Failed to update global_context entry : " + documentId;
                             logger.error(errorMessage, e);
-                            listener.onFailure(
-                                new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e))
-                            );
+                            listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
                         }
                     } else {
                         String errorMessage = "The template has already been provisioned so it can't be updated: " + documentId;
@@ -457,12 +446,14 @@ public <T> void doesTemplateExist(String documentId, Consumer<Boolean> booleanRe
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
             client.get(getRequest, ActionListener.wrap(response -> { booleanResultConsumer.accept(response.isExists()); }, exception -> {
                 context.restore();
-                logger.error("Failed to get template " + documentId, exception);
-                listener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
+                String errorMessage = "Failed to get template " + documentId;
+                logger.error(errorMessage);
+                listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(exception)));
             }));
         } catch (Exception e) {
-            logger.error("Failed to retrieve template from global context.", e);
-            listener.onFailure(new FlowFrameworkException(e.getMessage(), ExceptionsHelper.status(e)));
+            String errorMessage = "Failed to retrieve template from global context: " + documentId;
+            logger.error(errorMessage, e);
+            listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
         }
     }
 
@@ -490,17 +481,18 @@ public <T> void isWorkflowNotStarted(String documentId, Consumer<Boolean> boolea
                     WorkflowState workflowState = WorkflowState.parse(parser);
                     booleanResultConsumer.accept(workflowState.getProvisioningProgress().equals(ProvisioningProgress.NOT_STARTED.name()));
                 } catch (Exception e) {
-                    String message = "Failed to parse workflow state " + documentId;
-                    logger.error(message, e);
-                    listener.onFailure(new FlowFrameworkException(message, RestStatus.INTERNAL_SERVER_ERROR));
+                    String errorMessage = "Failed to parse workflow state " + documentId;
+                    logger.error(errorMessage, e);
+                    listener.onFailure(new FlowFrameworkException(errorMessage, RestStatus.INTERNAL_SERVER_ERROR));
                 }
             }, exception -> {
-                logger.error("Failed to get workflow state " + documentId, exception);
+                logger.error("Failed to get workflow state for {} ", documentId);
                 booleanResultConsumer.accept(false);
             }));
         } catch (Exception e) {
-            logger.error("Failed to retrieve workflow state to check provisioning status", e);
-            listener.onFailure(new FlowFrameworkException(e.getMessage(), ExceptionsHelper.status(e)));
+            String errorMessage = "Failed to retrieve workflow state to check provisioning status";
+            logger.error(errorMessage, e);
+            listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
         }
     }
 
@@ -516,9 +508,9 @@ public void updateFlowFrameworkSystemIndexDoc(
         ActionListener<UpdateResponse> listener
     ) {
         if (!doesIndexExist(WORKFLOW_STATE_INDEX)) {
-            String exceptionMessage = "Failed to update document for given workflow due to missing " + WORKFLOW_STATE_INDEX + " index";
-            logger.error(exceptionMessage);
-            listener.onFailure(new FlowFrameworkException(exceptionMessage, RestStatus.BAD_REQUEST));
+            String errorMessage = "Failed to update document for given workflow due to missing " + WORKFLOW_STATE_INDEX + " index";
+            logger.error(errorMessage);
+            listener.onFailure(new FlowFrameworkException(errorMessage, RestStatus.BAD_REQUEST));
         } else {
             try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
                 UpdateRequest updateRequest = new UpdateRequest(WORKFLOW_STATE_INDEX, documentId);
@@ -531,7 +523,7 @@ public void updateFlowFrameworkSystemIndexDoc(
             } catch (Exception e) {
                 String errorMessage = "Failed to update " + WORKFLOW_STATE_INDEX + " entry : " + documentId;
                 logger.error(errorMessage, e);
-                listener.onFailure(new FlowFrameworkException(errorMessage + " : " + e.getMessage(), ExceptionsHelper.status(e)));
+                listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
             }
         }
     }
@@ -550,9 +542,9 @@ public void updateFlowFrameworkSystemIndexDocWithScript(
         ActionListener<UpdateResponse> listener
     ) {
         if (!doesIndexExist(indexName)) {
-            String exceptionMessage = "Failed to update document for given workflow due to missing " + indexName + " index";
-            logger.error(exceptionMessage);
-            listener.onFailure(new Exception(exceptionMessage));
+            String errorMessage = "Failed to update document for given workflow due to missing " + indexName + " index";
+            logger.error(errorMessage);
+            listener.onFailure(new Exception(errorMessage));
         } else {
             try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
                 UpdateRequest updateRequest = new UpdateRequest(indexName, documentId);
@@ -563,10 +555,9 @@ public void updateFlowFrameworkSystemIndexDocWithScript(
                 // TODO: Implement our own concurrency control to improve on retry mechanism
                 client.update(updateRequest, ActionListener.runBefore(listener, context::restore));
             } catch (Exception e) {
-                logger.error("Failed to update {} entry : {}. {}", indexName, documentId, e.getMessage());
-                listener.onFailure(
-                    new FlowFrameworkException("Failed to update " + indexName + "entry: " + documentId, ExceptionsHelper.status(e))
-                );
+                String errorMessage = "Failed to update " + indexName + " entry : " + documentId;
+                logger.error(errorMessage, e);
+                listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
             }
         }
     }

src/main/java/org/opensearch/flowframework/model/Template.java

@@ -367,7 +367,7 @@ public String toJson() {
             XContentBuilder builder = JsonXContent.contentBuilder();
             return this.toXContent(builder, EMPTY_PARAMS).toString();
         } catch (IOException e) {
-            return "{\"error\": \"couldn't create JSON: " + e.getMessage() + "\"}";
+            return "{\"error\": \"couldn't create JSON from XContent\"}";
         }
     }
 
@@ -381,7 +381,7 @@ public String toYaml() {
             XContentBuilder builder = YamlXContent.contentBuilder();
             return this.toXContent(builder, EMPTY_PARAMS).toString();
         } catch (IOException e) {
-            return "error: couldn't create YAML: " + e.getMessage();
+            return "error: couldn't create YAML from XContent";
         }
     }
 

src/main/java/org/opensearch/flowframework/model/WorkflowStepValidator.java

@@ -8,8 +8,6 @@
  */
 package org.opensearch.flowframework.model;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.xcontent.ToXContentObject;
 import org.opensearch.core.xcontent.XContentBuilder;
@@ -22,8 +20,6 @@
  */
 public class WorkflowStepValidator implements ToXContentObject {
 
-    private static final Logger logger = LogManager.getLogger(WorkflowStepValidator.class);
-
     /** Inputs field name */
     private static final String INPUTS_FIELD = "inputs";
     /** Outputs field name */
@@ -33,28 +29,19 @@ public class WorkflowStepValidator implements ToXContentObject {
     /** Timeout field name */
     private static final String TIMEOUT = "timeout";
 
-    private String workflowStep;
     private List<String> inputs;
     private List<String> outputs;
     private List<String> requiredPlugins;
     private TimeValue timeout;
 
     /**
      * Instantiate the object representing a Workflow Step validator
-     * @param workflowStep name of the workflow step
      * @param inputs the workflow step inputs
      * @param outputs the workflow step outputs
      * @param requiredPlugins the required plugins for this workflow step
      * @param timeout the timeout for this workflow step
      */
-    public WorkflowStepValidator(
-        String workflowStep,
-        List<String> inputs,
-        List<String> outputs,
-        List<String> requiredPlugins,
-        TimeValue timeout
-    ) {
-        this.workflowStep = workflowStep;
+    public WorkflowStepValidator(List<String> inputs, List<String> outputs, List<String> requiredPlugins, TimeValue timeout) {
         this.inputs = inputs;
         this.outputs = outputs;
         this.requiredPlugins = requiredPlugins;

src/main/java/org/opensearch/flowframework/model/WorkflowValidator.java

@@ -40,7 +40,7 @@ public String toJson() {
             XContentBuilder builder = JsonXContent.contentBuilder();
             return this.toXContent(builder, EMPTY_PARAMS).toString();
         } catch (IOException e) {
-            return "{\"error\": \"couldn't create JSON: " + e.getMessage() + "\"}";
+            return "{\"error\": \"couldn't create JSON from XContent\"}";
         }
     }