gcp client for faaskeeper

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass

examples/example.py

@@ -1,7 +1,8 @@
 from faaskeeper.client import FaaSKeeperClient
 
+config = None
 try:
- client = FaaSKeeperClient("aws", "faaskeeper-dev", port=13001)
+ client = FaaSKeeperClient(config, 13001, True)
  client.start()
  #ret = client.create("/root/test2", b"test")
  #print(ret)

faaskeeper/client.py

@@ -20,6 +20,7 @@
  SetData,
 )
 from faaskeeper.providers.aws import AWSClient
+from faaskeeper.providers.gcp import GCPClient
 from faaskeeper.queue import (
  EventQueue,
  ResponseListener,
@@ -44,7 +45,7 @@ class FaaSKeeperClient:
  :param debug: full debug output of all operations
  """
 
- _providers = {CloudProvider.AWS: AWSClient}
+ _providers = {CloudProvider.AWS: AWSClient, CloudProvider.GCP: GCPClient}
 
  def __init__(
  self,

faaskeeper/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-
+from typing import Union
 
 class StorageType(Enum):
  PERSISTENT = 0
@@ -14,10 +14,11 @@ def deserialize(val: str) -> "StorageType":
 class QueueType(Enum):
  DYNAMODB = 0
  SQS = 1
+ PUBSUB = 2
 
  @staticmethod
  def deserialize(val: str) -> "QueueType":
- return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS}[val]
+ return {"dynamodb": QueueType.DYNAMODB, "sqs": QueueType.SQS, "pubsub": QueueType.PUBSUB}[val]
 
 
 class ClientChannel(Enum):
@@ -31,14 +32,15 @@ def deserialize(val: str) -> "ClientChannel":
 
 class CloudProvider(Enum):
  AWS = 0
+ GCP = 1
 
  @staticmethod
  def serialize(val: "CloudProvider") -> str:
- return {CloudProvider.AWS: "aws"}[val]
+ return {CloudProvider.AWS: "aws", CloudProvider.GCP: "gcp"}[val]
 
  @staticmethod
  def deserialize(val: str) -> "CloudProvider":
- return {"aws": CloudProvider.AWS}[val]
+ return {"aws": CloudProvider.AWS, "gcp": CloudProvider.GCP}[val]
 
 
 class AWSConfig:
@@ -55,6 +57,33 @@ def deserialize(data: dict) -> "AWSConfig":
  cfg._data_bucket = data["data-bucket"]
  return cfg
 
+class GCPConfig:
+ def __init__(self):
+ self._project_id: str
+ self._database: str
+ self._bucket_name: str
+
+ @property
+ def project_id(self) -> str:
+ return self._project_id
+
+ @property
+ def database(self) -> str:
+ return self._database
+
+ @property
+ def bucket_name(self) -> str:
+ return self._bucket_name
+
+ @staticmethod
+ def deserialize(data: dict) -> "GCPConfig":
+ cfg = GCPConfig()
+ cfg._project_id = data["project-id"]
+ cfg._database = data["database-name"]
+ deployment_name = data["deployment-name"]
+ bucket_name = data["bucket-name"]
+ cfg._bucket_name = f"sls-gcp-{deployment_name}-{bucket_name}"
+ return cfg
 
 class Config:
  def __init__(self):
@@ -65,7 +94,7 @@ def __init__(self):
  self._heartbeat_frequency: int
  self._user_storage: StorageType
  self._writer_queue: QueueType
- self._provider_cfg: AWSConfig
+ self._provider_cfg: Union[AWSConfig, GCPConfig]
  self._client_channel: ClientChannel
 
  @property
@@ -97,7 +126,7 @@ def writer_queue(self) -> QueueType:
  return self._writer_queue
 
  @property
- def provider_config(self) -> AWSConfig:
+ def provider_config(self) -> Union[AWSConfig, GCPConfig]:
  return self._provider_cfg
 
  @property
@@ -118,6 +147,9 @@ def deserialize(data: dict) -> "Config":
 
  if cfg._provider == CloudProvider.AWS:
  cfg._provider_cfg = AWSConfig.deserialize(data["aws"])
+ elif cfg._provider == CloudProvider.GCP:
+ data["gcp"]["deployment-name"] = cfg._deployment_name
+ cfg._provider_cfg = GCPConfig.deserialize(data["gcp"])
  else:
  raise NotImplementedError()
 

faaskeeper/providers/aws.py

@@ -54,7 +54,7 @@ def __init__(self, cfg: Config):
 
  def send_request(
  self,
- request_id: str,
+ request_id: str, # {session_id} - {local_idx}
  data: Dict[str, Union[str, bytes, int]],
  ):
  # FIXME: handle failure

faaskeeper/providers/gcp.py

@@ -0,0 +1,153 @@
+from typing import Dict, List, Optional, Tuple, Union
+from faaskeeper.config import Config, QueueType
+from faaskeeper.node import Node
+from faaskeeper.providers.provider import ProviderClient
+from faaskeeper.watch import Watch, WatchCallbackType, WatchType
+from faaskeeper.providers.serialization import S3Reader
+from faaskeeper.stats import StorageStatistics
+from faaskeeper.exceptions import NodeDoesntExistException, WatchSetFailureException
+from google.cloud import datastore
+from google.cloud import storage
+from google.cloud import pubsub_v1
+from datetime import datetime, timedelta
+
+import logging
+import json
+import google.cloud.exceptions
+import base64
+
+BENCHMARKING = True
+
+class GCPClient(ProviderClient):
+ def __init__(self, cfg: Config):
+ super().__init__(cfg)
+ self._log = logging.getLogger("GCPClient")
+ self._datastore = datastore.Client(project=self._config.provider_config.project_id, 
+ database=self._config.provider_config.database)
+ self._watch_table = f"faaskeeper-{self._config.deployment_name}-watch"
+ self._session_table = f"faaskeeper-{self._config.deployment_name}-users"
+ _storage_client = storage.Client()
+ self._bucket = _storage_client.bucket(self._config.provider_config.bucket_name)
+
+ # writer queue for now is pub/sub
+ if cfg.writer_queue == QueueType.PUBSUB:
+ batch_settings = pubsub_v1.types.BatchSettings(
+ max_messages=10, # default 100, now it is 10
+ max_bytes= 1 * 1000 * 1000, # default 1 MB, still 1 MB -> 1000 * 1000 KB
+ max_latency=0.01, # default 10 ms, now is 10ms
+ )
+ publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True) # enable FIFO
+ self._publisher_client = pubsub_v1.PublisherClient(publisher_options=publisher_options, batch_settings= batch_settings)
+
+ self._topic_id = "writer-queue-topic"
+ self._project_id = self._config.provider_config.project_id
+ self._topic_path = self._publisher_client.topic_path(self._project_id, self._topic_id)
+ else:
+ raise NotImplementedError()
+
+ def get_data_helper(self, path, include_data: bool, include_children: bool) -> Optional[Node]:
+ begin = datetime.now()
+ blob = self._bucket.get_blob(path)
+ if blob is not None:
+ file_content = blob.download_as_bytes()
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_read_time(int((end - begin) / timedelta(microseconds=1))) # 1 entity read
+ return S3Reader.deserialize(path, file_content, include_children, include_data)
+ return None
+
+ def get_data(self, path: str, watch_callback: Optional[WatchCallbackType], listen_address: Tuple[str, int]) -> Tuple[Node, Optional[Watch]]:
+ # read data
+ node = self.get_data_helper(path, True, True)
+ if node is not None:
+ watch: Optional[Watch] = None
+ if watch_callback is not None:
+ """
+ refer to the watch registration in aws.py
+ """
+ watch = self.register_watch(node, WatchType.GET_DATA, watch_callback, listen_address)
+ # second read
+ second_read_node = self.get_data_helper(path, True, True)
+ if second_read_node is None or node.modified.system != second_read_node.modified.system:
+ raise WatchSetFailureException(path)
+ # FIXME: insert watch into queue
+ return node, watch
+ else:
+ raise NodeDoesntExistException(path) 
+
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ begin = datetime.now()
+ if self._config.writer_queue == QueueType.PUBSUB:
+ data["timestamp"] = request_id
+ if "data" in data:
+ data["data"] = base64.b64encode(data["data"]).decode()
+ payload = json.dumps(data).encode()
+ _ = self._publisher_client.publish(
+ self._topic_path, payload, data["session_id"]
+ )
+
+ end = datetime.now()
+ if BENCHMARKING:
+ StorageStatistics.instance().add_write_time(int((end - begin) / timedelta(microseconds=1)))
+
+ def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, False, False)
+ # TODO: add watch
+ return node, None
+
+ def get_children(self, path: str, include_data: bool) -> Tuple[List[Node], Optional[Watch]]:
+ node = self.get_data_helper(path, include_data, True)
+ # TODO: add watch
+ return node, None
+
+ def register_session(self, session_id: str, sourceAddr: str, heartbeat: bool):
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._session_table, session_id)
+ node = datastore.Entity(key)
+ node.update({
+ "addr": sourceAddr,
+ "ephemerals": []
+ })
+
+ self._datastore.put(node)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering sessions on Datastore Database"
+ )
+
+ def register_watch(self, node: Node, watch_type: WatchType, watch: WatchCallbackType, listen_address: Tuple[str, int]) -> Watch:
+ item_names: Dict[WatchType, str] = {WatchType.GET_DATA: "getData"}
+
+ watch_name = item_names.get(watch_type)
+ assert watch_name
+ data_version = node.modified.system.sum
+ watch_data = [[data_version, listen_address[0], listen_address[1]]]
+
+ try:
+ with self._datastore.transaction():
+ key = self._datastore.key(self._watch_table, node.path)
+ node_info = self._datastore.get(key)
+ if node_info is not None:
+ if watch_name in node_info:
+ node_info[watch_name] = node_info[watch_name] + watch_data
+ else:
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ else:
+ # create one
+ node_info[watch_name] = watch_data
+ self._datastore.put(node_info)
+ StorageStatistics.instance().add_write_units(1)
+ except google.cloud.exceptions:
+ raise google.cloud.exceptions.BadRequest(
+ "Failure on registering the watch on Datastore Database"
+ ) 
+
+ return Watch(data_version, watch_type, watch)
+
+

faaskeeper/providers/provider.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, Tuple
-
+from typing import Dict, List, Optional, Tuple, Union
 from faaskeeper.config import Config
 from faaskeeper.node import Node
 from faaskeeper.operations import (
@@ -23,6 +22,14 @@ def get_data(
  ) -> Tuple[Node, Optional[Watch]]:
  pass
 
+ @abstractmethod
+ def send_request(
+ self,
+ request_id: str,
+ data: Dict[str, Union[str, bytes, int]],
+ ):
+ pass
+
  @abstractmethod
  def exists(self, path: str) -> Tuple[Optional[Node], Optional[Watch]]:
  pass