Skip to content

Activation id in header #3671

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rabbah merged 4 commits into from
Aug 22, 2018
Merged

Activation id in header #3671

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8a393f2
Apply review comments
Aug 14, 2018
5e313d1
Remove extra lines
Aug 15, 2018
46a8c10
Fix test failure
Aug 15, 2018
ec20733
Make ActivatioId value lowercased.
Aug 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 19 additions & 15 deletions core/controller/src/main/scala/whisk/core/controller/Actions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,24 +256,28 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
onComplete(invokeAction(user, actionWithMergedParams, payload, waitForResponse, cause = None)) {
case Success(Left(activationId)) =>
// non-blocking invoke or blocking invoke which got queued instead
complete(Accepted, activationId.toJsObject)
respondWithActivationIdHeader(activationId) {
complete(Accepted, activationId.toJsObject)
}
case Success(Right(activation)) =>
val response = if (result) activation.resultAsJson else activation.toExtendedJson

if (activation.response.isSuccess) {
complete(OK, response)
} else if (activation.response.isApplicationError) {
// actions that result is ApplicationError status are considered a 'success'
// and will have an 'error' property in the result - the HTTP status is OK
// and clients must check the response status if it exists
// NOTE: response status will not exist in the JSON object if ?result == true
// and instead clients must check if 'error' is in the JSON
// PRESERVING OLD BEHAVIOR and will address defect in separate change
complete(BadGateway, response)
} else if (activation.response.isContainerError) {
complete(BadGateway, response)
} else {
complete(InternalServerError, response)
respondWithActivationIdHeader(activation.activationId) {
if (activation.response.isSuccess) {
complete(OK, response)
} else if (activation.response.isApplicationError) {
// actions that result is ApplicationError status are considered a 'success'
// and will have an 'error' property in the result - the HTTP status is OK
// and clients must check the response status if it exists
// NOTE: response status will not exist in the JSON object if ?result == true
// and instead clients must check if 'error' is in the JSON
// PRESERVING OLD BEHAVIOR and will address defect in separate change
complete(BadGateway, response)
} else if (activation.response.isContainerError) {
complete(BadGateway, response)
} else {
complete(InternalServerError, response)
}
}
case Failure(t: RecordTooLargeException) =>
logging.debug(this, s"[POST] action payload was too large")
Expand Down
15 changes: 12 additions & 3 deletions core/controller/src/main/scala/whisk/core/controller/Entities.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@ package whisk.core.controller
import scala.concurrent.Future
import scala.language.postfixOps
import scala.util.Try

import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.model.StatusCodes.RequestEntityTooLarge
import akka.http.scaladsl.model.headers.RawHeader
import akka.http.scaladsl.server.Directive0
import akka.http.scaladsl.server.Directives
import akka.http.scaladsl.server.RequestContext
Expand Down Expand Up @@ -65,15 +65,24 @@ protected[controller] trait ValidateRequestSize extends Directives {
protected val fieldDescriptionForSizeError = "Request"
}

protected trait CustomHeaders extends Directives {
val ActivationIdHeader = "x-openwhisk-activation-id"

@rabbah rabbah Aug 17, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think @mhenke1 introduced a header with X-OW as the prefix. API gateway uses x-openwhisk. We should normalize and collocate these. Fwiw nginx also generates x-request-id header.

@style95 style95 Aug 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rabbah Agree.
How about using x-openwhisk?
Since x- header is a custom header specific to the application, it would be better to use the detailed one.

@style95 style95 Aug 20, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure changes to use the same OpenWhisk prefix for all custom headers should be included in this PR.


/** Add activation ID in headers */
protected def respondWithActivationIdHeader(activationId: ActivationId): Directive0 = {
respondWithHeader(RawHeader(ActivationIdHeader, activationId.asString))
}
}

/** A trait implementing the basic operations on WhiskEntities in support of the various APIs. */
trait WhiskCollectionAPI
extends Directives
with AuthenticatedRouteProvider
with AuthorizedRouteProvider
with ValidateRequestSize
with ReadOps
with WriteOps {

with WriteOps
with CustomHeaders {
/** The core collections require backend services to be injected in this trait. */
services: WhiskServices =>

Expand Down
4 changes: 3 additions & 1 deletion core/controller/src/main/scala/whisk/core/controller/Triggers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,9 @@ trait WhiskTriggersApi extends WhiskCollectionAPI {
.map { activation =>
activationStore.store(activation, context)
}
complete(Accepted, triggerActivationId.toJsObject)
respondWithActivationIdHeader(triggerActivationId) {
complete(Accepted, triggerActivationId.toJsObject)
}
} else {
logging
.debug(
Expand Down
50 changes: 27 additions & 23 deletions core/controller/src/main/scala/whisk/core/controller/WebActions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,7 +353,7 @@ protected[core] object WhiskWebActionsApi extends Directives {
headers.filter(_.lowercaseName != `Content-Type`.lowercaseName)
}

trait WhiskWebActionsApi extends Directives with ValidateRequestSize with PostActionActivation {
trait WhiskWebActionsApi extends Directives with ValidateRequestSize with PostActionActivation with CustomHeaders {
services: WhiskServices =>

/** API path invocation path for posting activations directly through the host. */
Expand Down Expand Up @@ -643,37 +643,41 @@ trait WhiskWebActionsApi extends Directives with ValidateRequestSize with PostAc
responseType: MediaExtension)(implicit transid: TransactionId) = {
onComplete(queuedActivation) {
case Success(Right(activation)) =>
val result = activation.resultAsJson
respondWithActivationIdHeader(activation.activationId) {
val result = activation.resultAsJson

if (activation.response.isSuccess || activation.response.isApplicationError) {
val resultPath = if (activation.response.isSuccess) {
projectResultField
} else {
// the activation produced an error response: therefore ignore
// the requested projection and unwrap the error instead
// and attempt to handle it per the desired response type (extension)
List(ActivationResponse.ERROR_FIELD)
}

if (activation.response.isSuccess || activation.response.isApplicationError) {
val resultPath = if (activation.response.isSuccess) {
projectResultField
val result = getFieldPath(activation.resultAsJson, resultPath)
result match {
case Some(projection) =>
val marshaler = Future(responseType.transcoder(projection, transid, webApiDirectives))
onComplete(marshaler) {
case Success(done) => done // all transcoders terminate the connection
case Failure(t) => terminate(InternalServerError)
}
case _ => terminate(NotFound, Messages.propertyNotFound)
}
} else {
// the activation produced an error response: therefore ignore
// the requested projection and unwrap the error instead
// and attempt to handle it per the desired response type (extension)
List(ActivationResponse.ERROR_FIELD)
}

val result = getFieldPath(activation.resultAsJson, resultPath)
result match {
case Some(projection) =>
val marshaler = Future(responseType.transcoder(projection, transid, webApiDirectives))
onComplete(marshaler) {
case Success(done) => done // all transcoders terminate the connection
case Failure(t) => terminate(InternalServerError)
}
case _ => terminate(NotFound, Messages.propertyNotFound)
terminate(BadRequest, Messages.errorProcessingRequest)
}
} else {
terminate(BadRequest, Messages.errorProcessingRequest)
}

case Success(Left(activationId)) =>
// blocking invoke which got queued instead
// this should not happen, instead it should be a blocking invoke timeout
logging.debug(this, "activation waiting period expired")
terminate(Accepted, Messages.responseNotReady)
respondWithActivationIdHeader(activationId) {
terminate(Accepted, Messages.responseNotReady)
}

case Failure(t: RejectRequest) => terminate(t.code, t.message)

Expand Down
12 changes: 12 additions & 0 deletions tests/src/test/scala/whisk/core/controller/test/ActionsApiTests.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import whisk.core.database.UserContext
import java.io.ByteArrayInputStream
import java.util.Base64

import akka.http.scaladsl.model.headers.RawHeader
import akka.stream.scaladsl._

/**
Expand Down Expand Up @@ -1211,13 +1212,15 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(Accepted)
val response = responseAs[JsObject]
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}

// it should "ignore &result when invoking nonblocking action"
Post(s"$collectionPath/${action.name}?result=true", args) ~> Route.seal(routes(creds)) ~> check {
status should be(Accepted)
val response = responseAs[JsObject]
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
}

Expand All @@ -1229,6 +1232,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(Accepted)
val response = responseAs[JsObject]
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
}

Expand Down Expand Up @@ -1263,6 +1267,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(Accepted)
val response = responseAs[JsObject]
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
}

Expand Down Expand Up @@ -1294,6 +1299,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(OK)
val response = responseAs[JsObject]
response should be(activation.resultAsJson)
headers should contain(RawHeader(ActivationIdHeader, activation.activationId.asString))
}
} finally {
deleteActivation(ActivationId(activation.docid.asString), context)
Expand Down Expand Up @@ -1363,14 +1369,18 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {

// will not wait long enough should get accepted status
Post(s"$collectionPath/${action.name}?blocking=true&timeout=100") ~> Route.seal(routes(creds)) ~> check {
val response = responseAs[JsObject]
status shouldBe Accepted
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))

}

// repeat this time wait longer than active ack delay
Post(s"$collectionPath/${action.name}?blocking=true&timeout=500") ~> Route.seal(routes(creds)) ~> check {
status shouldBe OK
val response = responseAs[JsObject]
response shouldBe activation.withoutLogs.toExtendedJson
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
} finally {
loadBalancer.whiskActivationStub = None
Expand Down Expand Up @@ -1398,6 +1408,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(InternalServerError)
val response = responseAs[JsObject]
response should be(activation.withoutLogs.toExtendedJson)
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
} finally {
deleteActivation(ActivationId(activation.docid.asString), context)
Expand All @@ -1412,6 +1423,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
status should be(Accepted)
val response = responseAs[JsObject]
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}
stream.toString should include(s"[WhiskActionMetaData] [GET] serving from datastore: ${CacheKey(action)}")
stream.reset()
Expand Down
6 changes: 3 additions & 3 deletions tests/src/test/scala/whisk/core/controller/test/ControllerTestCommon.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,7 @@ import whisk.common.TransactionId
import whisk.core.WhiskConfig
import whisk.core.connector.ActivationMessage
import whisk.core.containerpool.logging.LogStoreProvider
import whisk.core.controller.RestApiCommons
import whisk.core.controller.WhiskServices
import whisk.core.controller.{CustomHeaders, RestApiCommons, WhiskServices}
import whisk.core.database.{ActivationStoreProvider, CacheChangeNotification, DocumentFactory}
import whisk.core.database.test.DbUtils
import whisk.core.entitlement._
Expand All @@ -53,7 +52,8 @@ protected trait ControllerTestCommon
with DbUtils
with ExecHelpers
with WhiskServices
with StreamLogging {
with StreamLogging
with CustomHeaders {

val activeAckTopicIndex = ControllerInstanceId("0")

Expand Down
4 changes: 3 additions & 1 deletion tests/src/test/scala/whisk/core/controller/test/TriggersApiTests.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import org.scalatest.junit.JUnitRunner
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.server.Route
import akka.http.scaladsl.model.StatusCodes._

import akka.http.scaladsl.model.headers.RawHeader
import spray.json._
import spray.json.DefaultJsonProtocol._

Expand Down Expand Up @@ -369,6 +369,7 @@ class TriggersApiTests extends ControllerTestCommon with WhiskTriggersApi {
val JsString(id) = response.fields("activationId")
val activationId = ActivationId.parse(id).get
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))

val activationDoc = DocId(WhiskEntity.qualifiedName(namespace, activationId))
whisk.utils.retry({
Expand Down Expand Up @@ -399,6 +400,7 @@ class TriggersApiTests extends ControllerTestCommon with WhiskTriggersApi {
println(s"trying to delete async activation doc: '${activationDoc}'")
deleteActivation(ActivationId(activationDoc.asString), context)
response.fields("activationId") should not be None
headers should contain(RawHeader(ActivationIdHeader, response.fields("activationId").convertTo[String]))
}, 30, Some(1.second))
}
}
Expand Down
18 changes: 12 additions & 6 deletions tests/src/test/scala/whisk/core/controller/test/WebActionsApiTests.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -793,7 +793,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac
status should be(NoContent)
response.entity shouldBe HttpEntity.Empty
withClue(headers) {
headers.length shouldBe 0
headers.length shouldBe 1
headers.exists(_.is(ActivationIdHeader)) should be(true)
}
}

Expand Down Expand Up @@ -922,7 +923,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac

m(s"$testRoutePath/$path") ~> Route.seal(routes(creds)) ~> check {
status should be(statusCode)
headers shouldBe empty
headers.size shouldBe 1
headers.exists(_.is(ActivationIdHeader)) should be(true)
response.entity shouldBe HttpEntity.Empty
}
}
Expand All @@ -939,7 +941,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac

m(s"$testRoutePath/$path") ~> Route.seal(routes(creds)) ~> check {
status should be(statusCode)
headers shouldBe List(RawHeader("Set-Cookie", "a=b"))
headers should contain(RawHeader("Set-Cookie", "a=b"))
headers.exists(_.is(ActivationIdHeader)) should be(true)
response.entity shouldBe HttpEntity.Empty
}
}
Expand All @@ -959,7 +962,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac

m(s"$testRoutePath/$path") ~> addHeader("Accept", "application/json") ~> Route.seal(routes(creds)) ~> check {
status should be(statusCode)
headers shouldBe empty
headers.size shouldBe 1
headers.exists(_.is(ActivationIdHeader)) should be(true)
response.entity shouldBe HttpEntity.Empty
}
}
Expand All @@ -976,7 +980,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac

m(s"$testRoutePath/$path") ~> addHeader("Accept", "application/json") ~> Route.seal(routes(creds)) ~> check {
status should be(statusCode)
headers shouldBe List(RawHeader("Set-Cookie", "a=b"))
headers should contain(RawHeader("Set-Cookie", "a=b"))
headers.exists(_.is(ActivationIdHeader)) should be(true)
response.entity shouldBe HttpEntity.Empty
}
}
Expand Down Expand Up @@ -1165,7 +1170,8 @@ trait WebActionsApiBaseTests extends ControllerTestCommon with BeforeAndAfterEac
m(s"$testRoutePath/$path") ~> Route.seal(routes(creds)) ~> check {
status should be(OK)
mediaType shouldBe MediaTypes.`application/json`
headers shouldBe empty
headers.size shouldBe 1
headers.exists(_.is(ActivationIdHeader)) should be(true)
responseAs[String] shouldBe "hello world"
}
}
Expand Down