feedback

Author: Brendan Doyle

common/scala/src/main/resources/application.conf

@@ -438,13 +438,18 @@ whisk {
         #           uniqueName + displayName    253 (max pod name length in Kube)
         serdes-overhead = 6068 // 3034 bytes of metadata * 2 for extra headroom
 
+        # DEPRECATED, use store-blocking-result-level
         # Disables database store for blocking + successful activations
         # invocations made with `X-OW-EXTRA-LOGGING: on` header, will force the activation to be stored
         disable-store-result = false
 
-        # Disables database store for non-blocking + successful activations
+        # Result level to store in db for blocking activations (STORE_ALWAYS, STORE_FAILURES, STORE_FAILURES_NOT_APPLICATION_ERRORS)
         # invocations made with `X-OW-EXTRA-LOGGING: on` header, will force the activation to be stored
-        disable-store-non-blocking-result = false
+        store-blocking-result-level = "STORE_ALWAYS"
+
+        # Result level to store in db for blocking activations (STORE_ALWAYS, STORE_FAILURES, STORE_FAILURES_NOT_APPLICATION_ERRORS)
+        # invocations made with `X-OW-EXTRA-LOGGING: on` header, will force the activation to be stored
+        store-non-blocking-result-level = "STORE_ALWAYS"
 
         # Enable metadata logging of activations not stored in the database
         unstored-logs-enabled = false

common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationStore.scala

@@ -18,7 +18,6 @@
 package org.apache.openwhisk.core.database
 
 import java.time.Instant
-
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model.HttpRequest
 import spray.json.JsObject
@@ -33,9 +32,13 @@ import scala.concurrent.Future
 case class UserContext(user: Identity, request: HttpRequest = HttpRequest())
 
 trait ActivationStore {
-
+  /* DEPRECATED: disableStoreResult config is now deprecated replaced with blocking activation store level */
   protected val disableStoreResultConfig = loadConfigOrThrow[Boolean](ConfigKeys.disableStoreResult)
-  protected val disableStoreNonBlockingResultConfig = loadConfigOrThrow[Boolean](ConfigKeys.disableStoreNonBlockingResult)
+  protected val storeBlockingResultLevelConfig =
+    if (disableStoreResultConfig) ActivationStoreLevel.STORE_FAILURES
+    else ActivationStoreLevel.valueOf(loadConfigOrThrow[String](ConfigKeys.storeBlockingResultLevel))
+  protected val storeNonBlockingResultLevelConfig =
+    ActivationStoreLevel.valueOf(loadConfigOrThrow[String](ConfigKeys.storeNonBlockingResultLevel))
   protected val unstoredLogsEnabledConfig = loadConfigOrThrow[Boolean](ConfigKeys.unstoredLogsEnabled)
 
   /**
@@ -44,26 +47,26 @@ trait ActivationStore {
    * @param activation activation to store
    * @param isBlockingActivation is activation blocking
    * @param disableBlockingStore do not store activation if successful and blocking
-   * @param disableNonBlockingStore do not store activation if successful and non-blocking
+   * @param nonBlockingStoreLevel do not store activation if successful and non-blocking
    * @param context user and request context
    * @param transid transaction ID for request
    * @param notifier cache change notifier
    * @return Future containing DocInfo related to stored activation
    */
   def storeAfterCheck(activation: WhiskActivation,
                       isBlockingActivation: Boolean,
-                      disableBlockingStore: Option[Boolean],
-                      disableNonBlockingStore: Option[Boolean],
+                      blockingStoreLevel: Option[ActivationStoreLevel.Value],
+                      nonBlockingStoreLevel: Option[ActivationStoreLevel.Value],
                       context: UserContext)(implicit transid: TransactionId,
                                             notifier: Option[CacheChangeNotification],
                                             logging: Logging): Future[DocInfo] = {
     if (context.user.limits.storeActivations.getOrElse(true) &&
         shouldStoreActivation(
-          activation.response.isSuccess,
+          activation,
           isBlockingActivation,
           transid.meta.extraLogging,
-          disableBlockingStore.getOrElse(disableStoreResultConfig),
-          disableNonBlockingStore.getOrElse(disableStoreNonBlockingResultConfig))) {
+          blockingStoreLevel.getOrElse(storeBlockingResultLevelConfig),
+          nonBlockingStoreLevel.getOrElse(storeNonBlockingResultLevelConfig))) {
 
       store(activation, context)
     } else {
@@ -188,18 +191,29 @@ trait ActivationStore {
    * - an activation in debug mode
    * - activation stores is not disabled via a configuration parameter
    *
-   * @param isSuccess is successful activation
+   * @param activation to check
    * @param isBlocking is blocking activation
    * @param debugMode is logging header set to "on" for the invocation
-   * @param disableBlockingStore is disable store configured
+   * @param blockingStoreLevel level of activation status to store for blocking invocations
+   * @param nonBlockingStoreLevel level of activation status to store for blocking invocations
    * @return Should the activation be stored to the database
    */
-  private def shouldStoreActivation(isSuccess: Boolean,
+  private def shouldStoreActivation(activation: WhiskActivation,
                                     isBlocking: Boolean,
                                     debugMode: Boolean,
-                                    disableBlockingStore: Boolean,
-                                    disableNonBlockingStore: Boolean): Boolean = {
-    !isSuccess  || debugMode || (!isBlocking && !disableNonBlockingStore) || (isBlocking && !disableBlockingStore)
+                                    blockingStoreLevel: ActivationStoreLevel.Value,
+                                    nonBlockingStoreLevel: ActivationStoreLevel.Value): Boolean = {
+    def shouldStoreOnLevel(storageLevel: ActivationStoreLevel.Value): Boolean = {
+      storageLevel match {
+        case ActivationStoreLevel.STORE_ALWAYS   => true
+        case ActivationStoreLevel.STORE_FAILURES => !activation.response.isSuccess
+        case ActivationStoreLevel.STORE_FAILURES_NOT_APPLICATION_ERRORS =>
+          !activation.response.isSuccess && !activation.response.isApplicationError
+      }
+    }
+
+    debugMode || (isBlocking && shouldStoreOnLevel(blockingStoreLevel)) || (!isBlocking && shouldStoreOnLevel(
+      nonBlockingStoreLevel))
   }
 }
 

common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationStoreLevel.scala

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.openwhisk.core.database
+
+object ActivationStoreLevel extends Enumeration {
+  type ActivationStoreLevel = Value
+  val STORE_ALWAYS, STORE_FAILURES, STORE_FAILURES_NOT_APPLICATION_ERRORS = Value
+
+  def valueOf(value: String): Value = {
+    values
+      .find(_.toString == value.toUpperCase())
+      .getOrElse(throw new IllegalArgumentException(s"Invalid log level: $value"))
+  }
+}

core/controller/src/main/scala/org/apache/openwhisk/core/controller/actions/PrimitiveActions.scala

@@ -600,7 +600,10 @@ protected[actions] trait PrimitiveActions {
       }
     }
 
-    activationStore.storeAfterCheck(activation, blockingComposition, None, None, context)(transid, notifier = None, logging)
+    activationStore.storeAfterCheck(activation, blockingComposition, None, None, context)(
+      transid,
+      notifier = None,
+      logging)
 
     activation
   }

tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiWithDbPollingTests.scala

@@ -18,14 +18,13 @@
 package org.apache.openwhisk.core.controller.test
 
 import java.time.Instant
-
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.sprayJsonUnmarshaller
 import akka.http.scaladsl.model.StatusCodes._
 import akka.http.scaladsl.model.headers.RawHeader
 import akka.http.scaladsl.server.Route
 import org.apache.openwhisk.core.controller.WhiskActionsApi
 import org.apache.openwhisk.core.controller.actions.ControllerActivationConfig
-import org.apache.openwhisk.core.database.UserContext
+import org.apache.openwhisk.core.database.{ActivationStoreLevel, UserContext}
 import org.apache.openwhisk.core.entity._
 import org.junit.runner.RunWith
 import org.scalatest.junit.JUnitRunner
@@ -79,7 +78,7 @@ class ActionsApiWithDbPollingTests extends ControllerTestCommon with WhiskAction
     // storing the activation in the db will allow the db polling to retrieve it
     // the test harness makes sure the activation id observed by the test matches
     // the one generated by the api handler
-    storeActivation(activation, false, false, false, context)
+    storeActivation(activation, false, ActivationStoreLevel.STORE_ALWAYS, ActivationStoreLevel.STORE_ALWAYS, context)
     try {
       Post(s"$collectionPath/${action.name}?blocking=true") ~> Route.seal(routes(creds)) ~> check {
         status should be(OK)
@@ -114,7 +113,7 @@ class ActionsApiWithDbPollingTests extends ControllerTestCommon with WhiskAction
     // storing the activation in the db will allow the db polling to retrieve it
     // the test harness makes sure the activation id observed by the test matches
     // the one generated by the api handler
-    storeActivation(activation, false, false, false, context)
+    storeActivation(activation, false, ActivationStoreLevel.STORE_ALWAYS, ActivationStoreLevel.STORE_ALWAYS, context)
     try {
       Post(s"$collectionPath/${action.name}?blocking=true") ~> Route.seal(routes(creds)) ~> check {
         status should be(InternalServerError)