Forward header from a trigger to actions. (#5273)

* Forward header from a trigger to actions.

* Add the Apache header.

Author: style95

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Triggers.scala

@@ -18,7 +18,6 @@
 package org.apache.openwhisk.core.controller
 
 import java.time.{Clock, Instant}
-
 import scala.collection.immutable.Map
 import scala.concurrent.Future
 import scala.util.Try
@@ -28,7 +27,7 @@ import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
 import akka.http.scaladsl.model.HttpMethods.POST
 import akka.http.scaladsl.model.StatusCodes.{Accepted, BadRequest, InternalServerError, NoContent, OK, ServerError}
 import akka.http.scaladsl.model.Uri.Path
-import akka.http.scaladsl.model.headers.Authorization
+import akka.http.scaladsl.model.headers.{`Timeout-Access`, Authorization}
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.server.{RequestContext, RouteResult}
 import akka.http.scaladsl.unmarshalling.{Unmarshal, Unmarshaller}
@@ -151,7 +150,19 @@ trait WhiskTriggersApi extends WhiskCollectionAPI {
                 response = ActivationResponse.success(payload orElse Some(JsObject.empty)),
                 version = trigger.version,
                 duration = None)
-              val args: JsObject = trigger.parameters.merge(payload).getOrElse(JsObject.empty)
+              val headers = JsObject(
+                Map(new WebApiDirectives().headers -> request.headers
+                  .collect {
+                    case h if h.name != `Timeout-Access`.name => h.lowercaseName -> h.value
+                  }
+                  .toMap
+                  .toJson))
+
+              val mergedPayload = Some {
+                (headers.fields ++ (payload getOrElse JsObject.empty).fields).toJson.asJsObject
+              }
+
+              val args: JsObject = trigger.parameters.merge(mergedPayload).getOrElse(JsObject.empty)
 
               activateRules(user, args, trigger.rules.getOrElse(Map.empty))
                 .map(results => triggerActivation.withLogs(ActivationLogs(results.map(_.toJson.compactPrint).toVector)))

tests/dat/actions/params.js

@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+function main(args) {
+  return {args}
+}

tests/src/test/scala/org/apache/openwhisk/core/cli/test/WskRestBasicUsageTests.scala

@@ -21,9 +21,9 @@ import akka.http.scaladsl.model.StatusCodes.NotFound
 import akka.http.scaladsl.model.StatusCodes.OK
 import akka.http.scaladsl.model.StatusCodes.BadRequest
 import akka.http.scaladsl.model.StatusCodes.Conflict
+
 import java.time.Instant
 import java.time.Clock
-
 import scala.language.postfixOps
 import scala.concurrent.duration.DurationInt
 import scala.util.Random
@@ -36,22 +36,29 @@ import common.WhiskProperties
 import common.WskProps
 import common.WskTestHelpers
 import common.WskActorSystem
-import common.rest.WskRestOperations
+import common.rest.{RestResult, RunRestCmd, WskRestOperations}
 import spray.json.DefaultJsonProtocol._
 import spray.json._
 import org.apache.openwhisk.core.entity._
 import org.apache.openwhisk.core.entity.size.SizeInt
 import TestJsonArgs._
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.HttpMethods.POST
+import akka.http.scaladsl.model.{ContentTypes, HttpEntity, HttpHeader, HttpMethod, HttpRequest, HttpResponse, Uri}
+import akka.http.scaladsl.model.Uri.{Path, Query}
+import akka.http.scaladsl.model.headers.{Authorization, BasicHttpCredentials, RawHeader}
+import akka.util.ByteString
 import org.apache.openwhisk.http.Messages
 
 /**
  * Tests for basic CLI usage. Some of these tests require a deployed backend.
  */
 @RunWith(classOf[JUnitRunner])
-class WskRestBasicUsageTests extends TestHelpers with WskTestHelpers with WskActorSystem {
+class WskRestBasicUsageTests extends TestHelpers with WskTestHelpers with WskActorSystem with RunRestCmd {
 
   implicit val wskprops = WskProps()
-  val wsk = new WskRestOperations
+  implicit lazy override val executionContext = actorSystem.dispatcher
+  val wsk = new WskRestOperations()
   val defaultAction: Some[String] = Some(TestUtils.getTestActionFilename("hello.js"))
   val usrAgentHeaderRegEx: String = """\bUser-Agent\b": \[\s+"OpenWhisk\-CLI/1.\d+.*"""
 
@@ -737,4 +744,86 @@ class WskRestBasicUsageTests extends TestHelpers with WskTestHelpers with WskAct
       wsk.trigger.delete(triggerName).statusCode shouldBe OK
     }
   }
+
+  it should "forward headers as parameters to the associated action" in withAssetCleaner(wskprops) {
+    (wp, assetHelper) =>
+      val guestNamespace = wsk.namespace.whois()
+      val name = "triggerWithHeaders"
+      val actionName = "params"
+      val ruleName = "ruleWithHeaders"
+
+      assetHelper.withCleaner(wsk.trigger, name) { (trigger, _) =>
+        trigger.create(name)
+      }
+
+      assetHelper.withCleaner(wsk.action, actionName) { (action, _) =>
+        action.create(actionName, Some(TestUtils.getTestActionFilename("params.js")))
+      }
+
+      assetHelper.withCleaner(wsk.rule, ruleName) { (rule, _) =>
+        rule.create(ruleName, trigger = name, action = actionName)
+        rule.enable(ruleName)
+      }
+
+      val path = Path(s"$basePath/namespaces/$guestNamespace/triggers/$name")
+
+      val resp = requestEntityWithHeader(POST, path, List(RawHeader("Foo", "Bar")))(wp)
+      val result = new RestResult(resp.status.intValue, getTransactionId(resp), getRespData(resp))
+
+      withActivation(wsk.activation, result) { triggerActivation =>
+        val ruleActivation = triggerActivation.logs.get.map(_.parseJson.convertTo[common.RuleActivationResult]).head
+        withActivation(wsk.activation, ruleActivation.activationId) { actionActivation =>
+          actionActivation.response.result match {
+            case Some(result) =>
+              result.fields.get("args") map { headers =>
+                headers.asJsObject.fields.get("__ow_headers") map { params =>
+                  params.asJsObject.fields.get("foo") map { foo =>
+                    foo shouldBe JsString("Bar")
+                  }
+                }
+              }
+
+            case others =>
+              fail(s"no result found: $others")
+
+          }
+          actionActivation.cause shouldBe None
+        }
+      }
+  }
+
+  def requestEntityWithHeader(method: HttpMethod,
+                              path: Path,
+                              headers: List[HttpHeader],
+                              params: Map[String, String] = Map.empty,
+                              body: Option[String] = None)(implicit wp: WskProps): HttpResponse = {
+    val credentials = wp.authKey.split(":")
+    val creds = new BasicHttpCredentials(credentials(0), credentials(1))
+
+    // startsWith(http) includes https
+    val hostWithScheme = if (wp.apihost.startsWith("http")) {
+      Uri(wp.apihost)
+    } else {
+      Uri().withScheme("https").withHost(wp.apihost)
+    }
+
+    val request = HttpRequest(
+      method,
+      hostWithScheme.withPath(path).withQuery(Query(params)),
+      Authorization(creds) :: headers,
+      entity =
+        body.map(b => HttpEntity.Strict(ContentTypes.`application/json`, ByteString(b))).getOrElse(HttpEntity.Empty))
+    val response = Http().singleRequest(request, connectionContext).flatMap { _.toStrict(toStrictTimeout) }.futureValue
+
+    logger.debug(this, s"Request: $request")
+    logger.debug(this, s"Response: $response")
+
+    val validationErrors = validateRequestAndResponse(request, response)
+    if (validationErrors.nonEmpty) {
+      fail(
+        s"HTTP request or response did not match the Swagger spec.\nRequest: $request\n" +
+          s"Response: $response\nValidation Error: $validationErrors")
+    }
+    response
+  }
 }

tests/src/test/scala/system/basic/WskSequenceTests.scala

@@ -551,7 +551,14 @@ class WskSequenceTests extends TestHelpers with WskTestHelpers with StreamLoggin
     withActivation(wsk.activation, triggerFireRun) { triggerActivation =>
       val ruleActivation = triggerActivation.logs.get.map(_.parseJson.convertTo[RuleActivationResult]).head
       withActivation(wsk.activation, ruleActivation.activationId) { actionActivation =>
-        actionActivation.response.result shouldBe Some(triggerPayload)
+        actionActivation.response.result match {
+          case Some(result) =>
+            val (_, part2) = result.fields partition (p => p._1 == "__ow_headers") // excluding headers
+            JsObject(part2) shouldBe triggerPayload
+          case others =>
+            fail(s"no result found: $others")
+
+        }
         actionActivation.cause shouldBe None
       }
     }