We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support cloud events for KFServing custom framework 
/kind feature

**Describe the solution you'd like**
As a user when doing feature transformation the payload can be protobuf/avro or some other binary format delivered via HTTP or KAFKA in cloudevent binary format(https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#32-binary-content-mode) but currently KFServing custom framework [only supports JSON](https://github.com/kubeflow/kfserving/blob/master/python/kfserving/kfserving/handlers/http.py#L51).

- I'd like to send cloud event to KFServer via KAFKA in following format and KFServer pass the decoded cloud event to `preprocess` or `predict` handler
  
```
------------------ Message -------------------
Topic Name: mytopic
------------------- key ----------------------
Key: mykey
------------------ headers -------------------
ce_specversion: "1.0"
ce_type: "com.example.someevent"
ce_source: "/mycontext/subcontext"
ce_id: "1234-1234-1234"
ce_time: "2018-04-05T03:56:24Z"
content-type: application/protobuf
------------------- value --------------------
      ... application data encoded in protobuf ...
-----------------------------------------------

             consume                       http(cloudevent)
KAFKA ----------> Kafka Event Source -----------------> InferenceService

```




```python
def has_ce_headers(headers: typing.Dict[str, str]) -> bool:
    return (
        "ce-specversion" in headers
        and "ce-source" in headers
        and "ce-type" in headers
        and "ce-id" in headers
    )
    

 if has_ce_headers(self.request.headers):
        event = from_http(self.request.headers, self.request.body)
        body = event['data']
 else:
        try:
            body = json.loads(self.request.body)
        except json.decoder.JSONDecodeError as e:
            raise tornado.web.HTTPError(
                 status_code=HTTPStatus.BAD_REQUEST,
                 reason="Unrecognized request format: %s" % e)
 ```                   
                    
- Currently HTTP headers are not exposed to the `preprocess` or `predict` handler, if a cloud event is sent to KFServer it is not possible for user to check the http headers which can be useful for tracing and examining the original event source(kafka or http).
`async def predict(self, request: Dict, headers: Dict) -> Dict:`
`async def preprocess(self, request: Dict, headers: Dict) -> Dict:`

- Also it would be nice if KFServing can generate cloud event response so that it works with knative eventing channels seamlessly.

**Anything else you would like to add:**
[Miscellaneous information that will assist in solving the issue.]


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/kfserving/kfserving/handlers/http.py`
Content:
```
1 # Copyright 2020 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import inspect
16 import tornado.web
17 import json
18 from http import HTTPStatus
19 from kfserving.kfmodel_repository import KFModelRepository
20 
21 
22 class HTTPHandler(tornado.web.RequestHandler):
23     def initialize(self, models: KFModelRepository):
24         self.models = models  # pylint:disable=attribute-defined-outside-init
25 
26     def get_model(self, name: str):
27         model = self.models.get_model(name)
28         if model is None:
29             raise tornado.web.HTTPError(
30                 status_code=HTTPStatus.NOT_FOUND,
31                 reason="Model with name %s does not exist." % name
32             )
33         if not model.ready:
34             model.load()
35         return model
36 
37     def validate(self, request):
38         if ("instances" in request and not isinstance(request["instances"], list)) or \
39            ("inputs" in request and not isinstance(request["inputs"], list)):
40             raise tornado.web.HTTPError(
41                 status_code=HTTPStatus.BAD_REQUEST,
42                 reason="Expected \"instances\" or \"inputs\" to be a list"
43             )
44         return request
45 
46 
47 class PredictHandler(HTTPHandler):
48     async def post(self, name: str):
49         model = self.get_model(name)
50         try:
51             body = json.loads(self.request.body)
52         except json.decoder.JSONDecodeError as e:
53             raise tornado.web.HTTPError(
54                 status_code=HTTPStatus.BAD_REQUEST,
55                 reason="Unrecognized request format: %s" % e
56             )
57         request = model.preprocess(body)
58         request = self.validate(request)
59         response = (await model.predict(request)) if inspect.iscoroutinefunction(model.predict) else model.predict(request)
60         response = model.postprocess(response)
61         self.write(response)
62 
63 
64 class ExplainHandler(HTTPHandler):
65     async def post(self, name: str):
66         model = self.get_model(name)
67         try:
68             body = json.loads(self.request.body)
69         except json.decoder.JSONDecodeError as e:
70             raise tornado.web.HTTPError(
71                 status_code=HTTPStatus.BAD_REQUEST,
72                 reason="Unrecognized request format: %s" % e
73             )
74         request = model.preprocess(body)
75         request = self.validate(request)
76         response = (await model.explain(request)) if inspect.iscoroutinefunction(model.explain) else model.explain(request)
77         response = model.postprocess(response)
78         self.write(response)
79
```

Path: `python/kfserving/kfserving/kfmodel.py`
Content:
```
1 # Copyright 2020 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from typing import Dict
16 import sys
17 
18 import json
19 import tornado.web
20 from tornado.httpclient import AsyncHTTPClient
21 
22 PREDICTOR_URL_FORMAT = "http://{0}/v1/models/{1}:predict"
23 EXPLAINER_URL_FORMAT = "http://{0}/v1/models/{1}:explain"
24 PREDICTOR_V2_URL_FORMAT = "http://{0}/v2/models/{1}/infer"
25 EXPLAINER_V2_URL_FORMAT = "http://{0}/v2/models/{1}/explain"
26 
27 
28 # KFModel is intended to be subclassed by various components within KFServing.
29 class KFModel:
30 
31     def __init__(self, name: str):
32         self.name = name
33         self.ready = False
34         self.protocol = "v1"
35         self.predictor_host = None
36         self.explainer_host = None
37         # The timeout matches what is set in generated Istio resources.
38         # We generally don't want things to time out at the request level here,
39         # timeouts should be handled elsewhere in the system.
40         self.timeout = 600
41         self._http_client_instance = None
42 
43     @property
44     def _http_client(self):
45         if self._http_client_instance is None:
46             self._http_client_instance = AsyncHTTPClient(max_clients=sys.maxsize)
47         return self._http_client_instance
48 
49     def load(self) -> bool:
50         self.ready = True
51         return self.ready
52 
53     def preprocess(self, request: Dict) -> Dict:
54         return request
55 
56     def postprocess(self, request: Dict) -> Dict:
57         return request
58 
59     async def predict(self, request: Dict) -> Dict:
60         if not self.predictor_host:
61             raise NotImplementedError
62         predict_url = PREDICTOR_URL_FORMAT.format(self.predictor_host, self.name)
63         if self.protocol == "v2":
64             predict_url = PREDICTOR_V2_URL_FORMAT.format(self.predictor_host, self.name)
65         response = await self._http_client.fetch(
66             predict_url,
67             method='POST',
68             request_timeout=self.timeout,
69             body=json.dumps(request)
70         )
71         if response.code != 200:
72             raise tornado.web.HTTPError(
73                 status_code=response.code,
74                 reason=response.body)
75         return json.loads(response.body)
76 
77     async def explain(self, request: Dict) -> Dict:
78         if self.explainer_host is None:
79             raise NotImplementedError
80         explain_url = EXPLAINER_URL_FORMAT.format(self.predictor_host, self.name)
81         if self.protocol == "v2":
82             explain_url = EXPLAINER_V2_URL_FORMAT.format(self.predictor_host, self.name)
83         response = await self._http_client.fetch(
84             url=explain_url,
85             method='POST',
86             request_timeout=self.timeout,
87             body=json.dumps(request)
88         )
89         if response.code != 200:
90             raise tornado.web.HTTPError(
91                 status_code=response.code,
92                 reason=response.body)
93         return json.loads(response.body)
94
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/kfserving/kfserving/handlers/http.py b/python/kfserving/kfserving/handlers/http.py
--- a/python/kfserving/kfserving/handlers/http.py
+++ b/python/kfserving/kfserving/handlers/http.py
@@ -14,9 +14,15 @@
 
 import inspect
 import tornado.web
+import typing
 import json
+import pytz
+import cloudevents.exceptions as ce
+from cloudevents.http import CloudEvent, from_http, is_binary, is_structured, to_binary, to_structured
+from cloudevents.sdk.converters.util import has_binary_headers
 from http import HTTPStatus
 from kfserving.kfmodel_repository import KFModelRepository
+from datetime import datetime
 
 
 class HTTPHandler(tornado.web.RequestHandler):
@@ -43,21 +49,52 @@
             )
         return request
 
-
 class PredictHandler(HTTPHandler):
     async def post(self, name: str):
+        if has_binary_headers(self.request.headers):            
+            try:
+                #Use default unmarshaller if contenttype is set in header
+                if "ce-contenttype" in self.request.headers:
+                    body = from_http(self.request.headers, self.request.body)
+                else:
+                    body = from_http(self.request.headers, self.request.body, lambda x: x)
+            except (ce.MissingRequiredFields, ce.InvalidRequiredFields, ce.InvalidStructuredJSON, ce.InvalidHeadersFormat, ce.DataMarshallerError, ce.DataUnmarshallerError) as e:
+                raise tornado.web.HTTPError(
+                    status_code=HTTPStatus.BAD_REQUEST,
+                    reason="Cloud Event Exceptions: %s" % e
+                )
+        else:
+            try:
+                body = json.loads(self.request.body)
+            except json.decoder.JSONDecodeError as e:
+                raise tornado.web.HTTPError(
+                    status_code=HTTPStatus.BAD_REQUEST,
+                    reason="Unrecognized request format: %s" % e
+                )
+
         model = self.get_model(name)
-        try:
-            body = json.loads(self.request.body)
-        except json.decoder.JSONDecodeError as e:
-            raise tornado.web.HTTPError(
-                status_code=HTTPStatus.BAD_REQUEST,
-                reason="Unrecognized request format: %s" % e
-            )
         request = model.preprocess(body)
         request = self.validate(request)
         response = (await model.predict(request)) if inspect.iscoroutinefunction(model.predict) else model.predict(request)
         response = model.postprocess(response)
+
+        if has_binary_headers(self.request.headers):
+            event = CloudEvent(body._attributes, response)
+            if is_binary(self.request.headers):
+                eventheader, eventbody = to_binary(event)
+            elif is_structured(self.request.headers):
+                eventheader, eventbody = to_structured(event)
+            for k, v in eventheader.items():
+                if k != "ce-time":
+                    self.set_header(k, v)
+                else: #utc now() timestamp
+                    self.set_header('ce-time', datetime.utcnow().replace(tzinfo=pytz.utc).strftime('%Y-%m-%dT%H:%M:%S.%f%z'))
+
+            if isinstance(eventbody, (bytes, bytearray)):
+                response = eventbody
+            else:
+                response = eventbody.data
+
         self.write(response)
 
 
diff --git a/python/kfserving/kfserving/kfmodel.py b/python/kfserving/kfserving/kfmodel.py
--- a/python/kfserving/kfserving/kfmodel.py
+++ b/python/kfserving/kfserving/kfmodel.py
@@ -51,7 +51,16 @@
         return self.ready
 
     def preprocess(self, request: Dict) -> Dict:
-        return request
+        # If cloudevent dict, then parse 'data' field. Otherwise, pass through.
+        if  "data" in request \
+            and "time" in request \
+            and "type" in request \
+            and "source" in request \
+            and "id" in request \
+            and "specversion" in request:
+            return request["data"]
+        else:
+            return request
 
     def postprocess(self, request: Dict) -> Dict:
         return request
@@ -91,3 +100,4 @@
                 status_code=response.code,
                 reason=response.body)
         return json.loads(response.body)
+

{"golden_diff": "diff --git a/python/kfserving/kfserving/handlers/http.py b/python/kfserving/kfserving/handlers/http.py\n--- a/python/kfserving/kfserving/handlers/http.py\n+++ b/python/kfserving/kfserving/handlers/http.py\n@@ -14,9 +14,15 @@\n \n import inspect\n import tornado.web\n+import typing\n import json\n+import pytz\n+import cloudevents.exceptions as ce\n+from cloudevents.http import CloudEvent, from_http, is_binary, is_structured, to_binary, to_structured\n+from cloudevents.sdk.converters.util import has_binary_headers\n from http import HTTPStatus\n from kfserving.kfmodel_repository import KFModelRepository\n+from datetime import datetime\n \n \n class HTTPHandler(tornado.web.RequestHandler):\n@@ -43,21 +49,52 @@\n             )\n         return request\n \n-\n class PredictHandler(HTTPHandler):\n     async def post(self, name: str):\n+        if has_binary_headers(self.request.headers):            \n+            try:\n+                #Use default unmarshaller if contenttype is set in header\n+                if \"ce-contenttype\" in self.request.headers:\n+                    body = from_http(self.request.headers, self.request.body)\n+                else:\n+                    body = from_http(self.request.headers, self.request.body, lambda x: x)\n+            except (ce.MissingRequiredFields, ce.InvalidRequiredFields, ce.InvalidStructuredJSON, ce.InvalidHeadersFormat, ce.DataMarshallerError, ce.DataUnmarshallerError) as e:\n+                raise tornado.web.HTTPError(\n+                    status_code=HTTPStatus.BAD_REQUEST,\n+                    reason=\"Cloud Event Exceptions: %s\" % e\n+                )\n+        else:\n+            try:\n+                body = json.loads(self.request.body)\n+            except json.decoder.JSONDecodeError as e:\n+                raise tornado.web.HTTPError(\n+                    status_code=HTTPStatus.BAD_REQUEST,\n+                    reason=\"Unrecognized request format: %s\" % e\n+                )\n+\n         model = self.get_model(name)\n-        try:\n-            body = json.loads(self.request.body)\n-        except json.decoder.JSONDecodeError as e:\n-            raise tornado.web.HTTPError(\n-                status_code=HTTPStatus.BAD_REQUEST,\n-                reason=\"Unrecognized request format: %s\" % e\n-            )\n         request = model.preprocess(body)\n         request = self.validate(request)\n         response = (await model.predict(request)) if inspect.iscoroutinefunction(model.predict) else model.predict(request)\n         response = model.postprocess(response)\n+\n+        if has_binary_headers(self.request.headers):\n+            event = CloudEvent(body._attributes, response)\n+            if is_binary(self.request.headers):\n+                eventheader, eventbody = to_binary(event)\n+            elif is_structured(self.request.headers):\n+                eventheader, eventbody = to_structured(event)\n+            for k, v in eventheader.items():\n+                if k != \"ce-time\":\n+                    self.set_header(k, v)\n+                else: #utc now() timestamp\n+                    self.set_header('ce-time', datetime.utcnow().replace(tzinfo=pytz.utc).strftime('%Y-%m-%dT%H:%M:%S.%f%z'))\n+\n+            if isinstance(eventbody, (bytes, bytearray)):\n+                response = eventbody\n+            else:\n+                response = eventbody.data\n+\n         self.write(response)\n \n \ndiff --git a/python/kfserving/kfserving/kfmodel.py b/python/kfserving/kfserving/kfmodel.py\n--- a/python/kfserving/kfserving/kfmodel.py\n+++ b/python/kfserving/kfserving/kfmodel.py\n@@ -51,7 +51,16 @@\n         return self.ready\n \n     def preprocess(self, request: Dict) -> Dict:\n-        return request\n+        # If cloudevent dict, then parse 'data' field. Otherwise, pass through.\n+        if  \"data\" in request \\\n+            and \"time\" in request \\\n+            and \"type\" in request \\\n+            and \"source\" in request \\\n+            and \"id\" in request \\\n+            and \"specversion\" in request:\n+            return request[\"data\"]\n+        else:\n+            return request\n \n     def postprocess(self, request: Dict) -> Dict:\n         return request\n@@ -91,3 +100,4 @@\n                 status_code=response.code,\n                 reason=response.body)\n         return json.loads(response.body)\n+\n", "issue": "Support cloud events for KFServing custom framework \n/kind feature\r\n\r\n**Describe the solution you'd like**\r\nAs a user when doing feature transformation the payload can be protobuf/avro or some other binary format delivered via HTTP or KAFKA in cloudevent binary format(https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#32-binary-content-mode) but currently KFServing custom framework [only supports JSON](https://github.com/kubeflow/kfserving/blob/master/python/kfserving/kfserving/handlers/http.py#L51).\r\n\r\n- I'd like to send cloud event to KFServer via KAFKA in following format and KFServer pass the decoded cloud event to `preprocess` or `predict` handler\r\n  \r\n```\r\n------------------ Message -------------------\r\nTopic Name: mytopic\r\n------------------- key ----------------------\r\nKey: mykey\r\n------------------ headers -------------------\r\nce_specversion: \"1.0\"\r\nce_type: \"com.example.someevent\"\r\nce_source: \"/mycontext/subcontext\"\r\nce_id: \"1234-1234-1234\"\r\nce_time: \"2018-04-05T03:56:24Z\"\r\ncontent-type: application/protobuf\r\n------------------- value --------------------\r\n      ... application data encoded in protobuf ...\r\n-----------------------------------------------\r\n\r\n             consume                       http(cloudevent)\r\nKAFKA ----------> Kafka Event Source -----------------> InferenceService\r\n\r\n```\r\n\r\n\r\n\r\n\r\n```python\r\ndef has_ce_headers(headers: typing.Dict[str, str]) -> bool:\r\n    return (\r\n        \"ce-specversion\" in headers\r\n        and \"ce-source\" in headers\r\n        and \"ce-type\" in headers\r\n        and \"ce-id\" in headers\r\n    )\r\n    \r\n\r\n if has_ce_headers(self.request.headers):\r\n        event = from_http(self.request.headers, self.request.body)\r\n        body = event['data']\r\n else:\r\n        try:\r\n            body = json.loads(self.request.body)\r\n        except json.decoder.JSONDecodeError as e:\r\n            raise tornado.web.HTTPError(\r\n                 status_code=HTTPStatus.BAD_REQUEST,\r\n                 reason=\"Unrecognized request format: %s\" % e)\r\n ```                   \r\n                    \r\n- Currently HTTP headers are not exposed to the `preprocess` or `predict` handler, if a cloud event is sent to KFServer it is not possible for user to check the http headers which can be useful for tracing and examining the original event source(kafka or http).\r\n`async def predict(self, request: Dict, headers: Dict) -> Dict:`\r\n`async def preprocess(self, request: Dict, headers: Dict) -> Dict:`\r\n\r\n- Also it would be nice if KFServing can generate cloud event response so that it works with knative eventing channels seamlessly.\r\n\r\n**Anything else you would like to add:**\r\n[Miscellaneous information that will assist in solving the issue.]\r\n\n", "before_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport inspect\nimport tornado.web\nimport json\nfrom http import HTTPStatus\nfrom kfserving.kfmodel_repository import KFModelRepository\n\n\nclass HTTPHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get_model(self, name: str):\n        model = self.models.get_model(name)\n        if model is None:\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.NOT_FOUND,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n        if not model.ready:\n            model.load()\n        return model\n\n    def validate(self, request):\n        if (\"instances\" in request and not isinstance(request[\"instances\"], list)) or \\\n           (\"inputs\" in request and not isinstance(request[\"inputs\"], list)):\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.BAD_REQUEST,\n                reason=\"Expected \\\"instances\\\" or \\\"inputs\\\" to be a list\"\n            )\n        return request\n\n\nclass PredictHandler(HTTPHandler):\n    async def post(self, name: str):\n        model = self.get_model(name)\n        try:\n            body = json.loads(self.request.body)\n        except json.decoder.JSONDecodeError as e:\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.BAD_REQUEST,\n                reason=\"Unrecognized request format: %s\" % e\n            )\n        request = model.preprocess(body)\n        request = self.validate(request)\n        response = (await model.predict(request)) if inspect.iscoroutinefunction(model.predict) else model.predict(request)\n        response = model.postprocess(response)\n        self.write(response)\n\n\nclass ExplainHandler(HTTPHandler):\n    async def post(self, name: str):\n        model = self.get_model(name)\n        try:\n            body = json.loads(self.request.body)\n        except json.decoder.JSONDecodeError as e:\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.BAD_REQUEST,\n                reason=\"Unrecognized request format: %s\" % e\n            )\n        request = model.preprocess(body)\n        request = self.validate(request)\n        response = (await model.explain(request)) if inspect.iscoroutinefunction(model.explain) else model.explain(request)\n        response = model.postprocess(response)\n        self.write(response)\n", "path": "python/kfserving/kfserving/handlers/http.py"}, {"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\nimport sys\n\nimport json\nimport tornado.web\nfrom tornado.httpclient import AsyncHTTPClient\n\nPREDICTOR_URL_FORMAT = \"http://{0}/v1/models/{1}:predict\"\nEXPLAINER_URL_FORMAT = \"http://{0}/v1/models/{1}:explain\"\nPREDICTOR_V2_URL_FORMAT = \"http://{0}/v2/models/{1}/infer\"\nEXPLAINER_V2_URL_FORMAT = \"http://{0}/v2/models/{1}/explain\"\n\n\n# KFModel is intended to be subclassed by various components within KFServing.\nclass KFModel:\n\n    def __init__(self, name: str):\n        self.name = name\n        self.ready = False\n        self.protocol = \"v1\"\n        self.predictor_host = None\n        self.explainer_host = None\n        # The timeout matches what is set in generated Istio resources.\n        # We generally don't want things to time out at the request level here,\n        # timeouts should be handled elsewhere in the system.\n        self.timeout = 600\n        self._http_client_instance = None\n\n    @property\n    def _http_client(self):\n        if self._http_client_instance is None:\n            self._http_client_instance = AsyncHTTPClient(max_clients=sys.maxsize)\n        return self._http_client_instance\n\n    def load(self) -> bool:\n        self.ready = True\n        return self.ready\n\n    def preprocess(self, request: Dict) -> Dict:\n        return request\n\n    def postprocess(self, request: Dict) -> Dict:\n        return request\n\n    async def predict(self, request: Dict) -> Dict:\n        if not self.predictor_host:\n            raise NotImplementedError\n        predict_url = PREDICTOR_URL_FORMAT.format(self.predictor_host, self.name)\n        if self.protocol == \"v2\":\n            predict_url = PREDICTOR_V2_URL_FORMAT.format(self.predictor_host, self.name)\n        response = await self._http_client.fetch(\n            predict_url,\n            method='POST',\n            request_timeout=self.timeout,\n            body=json.dumps(request)\n        )\n        if response.code != 200:\n            raise tornado.web.HTTPError(\n                status_code=response.code,\n                reason=response.body)\n        return json.loads(response.body)\n\n    async def explain(self, request: Dict) -> Dict:\n        if self.explainer_host is None:\n            raise NotImplementedError\n        explain_url = EXPLAINER_URL_FORMAT.format(self.predictor_host, self.name)\n        if self.protocol == \"v2\":\n            explain_url = EXPLAINER_V2_URL_FORMAT.format(self.predictor_host, self.name)\n        response = await self._http_client.fetch(\n            url=explain_url,\n            method='POST',\n            request_timeout=self.timeout,\n            body=json.dumps(request)\n        )\n        if response.code != 200:\n            raise tornado.web.HTTPError(\n                status_code=response.code,\n                reason=response.body)\n        return json.loads(response.body)\n", "path": "python/kfserving/kfserving/kfmodel.py"}], "after_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport inspect\nimport tornado.web\nimport typing\nimport json\nimport pytz\nimport cloudevents.exceptions as ce\nfrom cloudevents.http import CloudEvent, from_http, is_binary, is_structured, to_binary, to_structured\nfrom cloudevents.sdk.converters.util import has_binary_headers\nfrom http import HTTPStatus\nfrom kfserving.kfmodel_repository import KFModelRepository\nfrom datetime import datetime\n\n\nclass HTTPHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get_model(self, name: str):\n        model = self.models.get_model(name)\n        if model is None:\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.NOT_FOUND,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n        if not model.ready:\n            model.load()\n        return model\n\n    def validate(self, request):\n        if (\"instances\" in request and not isinstance(request[\"instances\"], list)) or \\\n           (\"inputs\" in request and not isinstance(request[\"inputs\"], list)):\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.BAD_REQUEST,\n                reason=\"Expected \\\"instances\\\" or \\\"inputs\\\" to be a list\"\n            )\n        return request\n\nclass PredictHandler(HTTPHandler):\n    async def post(self, name: str):\n        if has_binary_headers(self.request.headers):            \n            try:\n                #Use default unmarshaller if contenttype is set in header\n                if \"ce-contenttype\" in self.request.headers:\n                    body = from_http(self.request.headers, self.request.body)\n                else:\n                    body = from_http(self.request.headers, self.request.body, lambda x: x)\n            except (ce.MissingRequiredFields, ce.InvalidRequiredFields, ce.InvalidStructuredJSON, ce.InvalidHeadersFormat, ce.DataMarshallerError, ce.DataUnmarshallerError) as e:\n                raise tornado.web.HTTPError(\n                    status_code=HTTPStatus.BAD_REQUEST,\n                    reason=\"Cloud Event Exceptions: %s\" % e\n                )\n        else:\n            try:\n                body = json.loads(self.request.body)\n            except json.decoder.JSONDecodeError as e:\n                raise tornado.web.HTTPError(\n                    status_code=HTTPStatus.BAD_REQUEST,\n                    reason=\"Unrecognized request format: %s\" % e\n                )\n\n        model = self.get_model(name)\n        request = model.preprocess(body)\n        request = self.validate(request)\n        response = (await model.predict(request)) if inspect.iscoroutinefunction(model.predict) else model.predict(request)\n        response = model.postprocess(response)\n\n        if has_binary_headers(self.request.headers):\n            event = CloudEvent(body._attributes, response)\n            if is_binary(self.request.headers):\n                eventheader, eventbody = to_binary(event)\n            elif is_structured(self.request.headers):\n                eventheader, eventbody = to_structured(event)\n            for k, v in eventheader.items():\n                if k != \"ce-time\":\n                    self.set_header(k, v)\n                else: #utc now() timestamp\n                    self.set_header('ce-time', datetime.utcnow().replace(tzinfo=pytz.utc).strftime('%Y-%m-%dT%H:%M:%S.%f%z'))\n\n            if isinstance(eventbody, (bytes, bytearray)):\n                response = eventbody\n            else:\n                response = eventbody.data\n\n        self.write(response)\n\n\nclass ExplainHandler(HTTPHandler):\n    async def post(self, name: str):\n        model = self.get_model(name)\n        try:\n            body = json.loads(self.request.body)\n        except json.decoder.JSONDecodeError as e:\n            raise tornado.web.HTTPError(\n                status_code=HTTPStatus.BAD_REQUEST,\n                reason=\"Unrecognized request format: %s\" % e\n            )\n        request = model.preprocess(body)\n        request = self.validate(request)\n        response = (await model.explain(request)) if inspect.iscoroutinefunction(model.explain) else model.explain(request)\n        response = model.postprocess(response)\n        self.write(response)\n", "path": "python/kfserving/kfserving/handlers/http.py"}, {"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\nimport sys\n\nimport json\nimport tornado.web\nfrom tornado.httpclient import AsyncHTTPClient\n\nPREDICTOR_URL_FORMAT = \"http://{0}/v1/models/{1}:predict\"\nEXPLAINER_URL_FORMAT = \"http://{0}/v1/models/{1}:explain\"\nPREDICTOR_V2_URL_FORMAT = \"http://{0}/v2/models/{1}/infer\"\nEXPLAINER_V2_URL_FORMAT = \"http://{0}/v2/models/{1}/explain\"\n\n\n# KFModel is intended to be subclassed by various components within KFServing.\nclass KFModel:\n\n    def __init__(self, name: str):\n        self.name = name\n        self.ready = False\n        self.protocol = \"v1\"\n        self.predictor_host = None\n        self.explainer_host = None\n        # The timeout matches what is set in generated Istio resources.\n        # We generally don't want things to time out at the request level here,\n        # timeouts should be handled elsewhere in the system.\n        self.timeout = 600\n        self._http_client_instance = None\n\n    @property\n    def _http_client(self):\n        if self._http_client_instance is None:\n            self._http_client_instance = AsyncHTTPClient(max_clients=sys.maxsize)\n        return self._http_client_instance\n\n    def load(self) -> bool:\n        self.ready = True\n        return self.ready\n\n    def preprocess(self, request: Dict) -> Dict:\n        # If cloudevent dict, then parse 'data' field. Otherwise, pass through.\n        if  \"data\" in request \\\n            and \"time\" in request \\\n            and \"type\" in request \\\n            and \"source\" in request \\\n            and \"id\" in request \\\n            and \"specversion\" in request:\n            return request[\"data\"]\n        else:\n            return request\n\n    def postprocess(self, request: Dict) -> Dict:\n        return request\n\n    async def predict(self, request: Dict) -> Dict:\n        if not self.predictor_host:\n            raise NotImplementedError\n        predict_url = PREDICTOR_URL_FORMAT.format(self.predictor_host, self.name)\n        if self.protocol == \"v2\":\n            predict_url = PREDICTOR_V2_URL_FORMAT.format(self.predictor_host, self.name)\n        response = await self._http_client.fetch(\n            predict_url,\n            method='POST',\n            request_timeout=self.timeout,\n            body=json.dumps(request)\n        )\n        if response.code != 200:\n            raise tornado.web.HTTPError(\n                status_code=response.code,\n                reason=response.body)\n        return json.loads(response.body)\n\n    async def explain(self, request: Dict) -> Dict:\n        if self.explainer_host is None:\n            raise NotImplementedError\n        explain_url = EXPLAINER_URL_FORMAT.format(self.predictor_host, self.name)\n        if self.protocol == \"v2\":\n            explain_url = EXPLAINER_V2_URL_FORMAT.format(self.predictor_host, self.name)\n        response = await self._http_client.fetch(\n            url=explain_url,\n            method='POST',\n            request_timeout=self.timeout,\n            body=json.dumps(request)\n        )\n        if response.code != 200:\n            raise tornado.web.HTTPError(\n                status_code=response.code,\n                reason=response.body)\n        return json.loads(response.body)\n\n", "path": "python/kfserving/kfserving/kfmodel.py"}]}