diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,13 @@ +# Azure Functions Haskell Worker + +**[NOTE]** This is a very early work in progress. + +Azure Functions Haskell Worker is a __library__ that runs Azure functions inside the worker process. +This means that a "real" functions application (executable) is meant to use the worker library +and run its `main` as `Azure.Functions.Worker.runWorker`. + +Worker process provides two main commands: + +- [run](src/Azure/Functions/Commands/Run.hs) is a main entry point that starts the worker and funs functions +- [init](src/Azure/Functions/Commands/Init.hs) is a command that allows the worker process to configure itself for the Azure Functions Host. Its main responsibility is to write `worker.config.json`. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,5 @@ +import Data.ProtoLens.Setup +main = defaultMainGeneratingProtos "protobuf/src/proto" + +-- import Distribution.Simple +-- main = defaultMain diff --git a/azure-functions-worker.cabal b/azure-functions-worker.cabal new file mode 100644 --- /dev/null +++ b/azure-functions-worker.cabal @@ -0,0 +1,87 @@ +cabal-version: 2.4 +name: azure-functions-worker +version: 0.0.0.0 +synopsis: Azure Functions Worker +description: Azure Functions Worker. +license: MIT +author: Alexey Raga +maintainer: Alexey Raga +category: Azure, Serverless, Cloud +extra-source-files: README.md + protobuf/src/proto/**/*.proto + +custom-setup + setup-depends: + Cabal + , base + , proto-lens-setup + +common config + default-language: Haskell2010 + -- ghc-options: -Wall -Werror + +library + import: config + hs-source-dirs: src + exposed-modules: Azure.Functions.Contract + Azure.Functions.Worker + Azure.Functions.Registry + Azure.Functions.Function + Azure.Functions.Bindings.Class + Azure.Functions.Bindings.Blob + Azure.Functions.Bindings.HTTP + Azure.Functions.Bindings.ServiceBus + + autogen-modules: Proto.FunctionRpc + Proto.FunctionRpc_Fields + Proto.Identity.ClaimsIdentityRpc + Proto.Shared.NullableTypes + Paths_azure_functions_worker + + other-modules: Azure.Functions.Bindings.Shared + Azure.Functions.Commands.Init + Azure.Functions.Commands.Run + Azure.Functions.Internal.Lens + Azure.Functions.Internal.Runtime + Azure.Functions.Internal.Templates + Proto.FunctionRpc + Proto.FunctionRpc_Fields + Proto.FunctionRpc_Helpers + Proto.Identity.ClaimsIdentityRpc + Proto.Shared.NullableTypes + Paths_azure_functions_worker + + build-tool-depends: proto-lens-protoc:proto-lens-protoc >= 0.6 && < 0.7 + + build-depends: base >= 4.12 && < 4.13 + , aeson + , async + , bytestring + , containers + , directory + , filepath + , glabrous >= 2.0.2 && < 3 + , http2-grpc-proto-lens >= 0.1 && < 0.2 + , http2-client-grpc >= 0.8 && < 0.9 + , http2-grpc-types + , http2-client >= 0.9 && < 0.10 + , http-types >= 0.12 && < 0.13 + , lens-family >= 2.0 && < 3 + , lens-family-core >= 2.0 && < 3 + , mtl + , network-uri >= 2.6.3 && < 2.7 + , optparse-applicative >= 0.15.1 && < 0.16 + , proto-lens-runtime >= 0.6 && < 0.7 + , proto-lens-protobuf-types >= 0.6 && < 0.7 + , raw-strings-qq >= 1.1 && < 2 + , stm + , text + , time + +test-suite azure-functions-test + import: config + hs-source-dirs: test + type: exitcode-stdio-1.0 + main-is: MyLibTest.hs + build-depends: base >=4.12 && <4.13 + , azure-functions-worker diff --git a/protobuf/src/proto/FunctionRpc.proto b/protobuf/src/proto/FunctionRpc.proto new file mode 100644 --- /dev/null +++ b/protobuf/src/proto/FunctionRpc.proto @@ -0,0 +1,484 @@ +syntax = "proto3"; +// protobuf vscode extension: https://marketplace.visualstudio.com/items?itemName=zxh404.vscode-proto3 + +option java_multiple_files = true; +option java_package = "com.microsoft.azure.functions.rpc.messages"; +option java_outer_classname = "FunctionProto"; +option csharp_namespace = "Microsoft.Azure.WebJobs.Script.Grpc.Messages"; +option go_package ="github.com/Azure/azure-functions-go-worker/internal/rpc"; + +package AzureFunctionsRpcMessages; + +import "google/protobuf/duration.proto"; +import "identity/ClaimsIdentityRpc.proto"; +import "shared/NullableTypes.proto"; + +// Interface exported by the server. +service FunctionRpc { + rpc EventStream (stream StreamingMessage) returns (stream StreamingMessage) {} +} + +message StreamingMessage { + // Used to identify message between host and worker + string request_id = 1; + + // Payload of the message + oneof content { + + // Worker initiates stream + StartStream start_stream = 20; + + // Host sends capabilities/init data to worker + WorkerInitRequest worker_init_request = 17; + // Worker responds after initializing with its capabilities & status + WorkerInitResponse worker_init_response = 16; + + // Worker periodically sends empty heartbeat message to host + WorkerHeartbeat worker_heartbeat = 15; + + // Host sends terminate message to worker. + // Worker terminates if it can, otherwise host terminates after a grace period + WorkerTerminate worker_terminate = 14; + + // Add any worker relevant status to response + WorkerStatusRequest worker_status_request = 12; + WorkerStatusResponse worker_status_response = 13; + + // On file change event, host sends notification to worker + FileChangeEventRequest file_change_event_request = 6; + + // Worker requests a desired action (restart worker, reload function) + WorkerActionResponse worker_action_response = 7; + + // Host sends required metadata to worker to load function + FunctionLoadRequest function_load_request = 8; + // Worker responds after loading with the load result + FunctionLoadResponse function_load_response = 9; + + // Host requests a given invocation + InvocationRequest invocation_request = 4; + + // Worker responds to a given invocation + InvocationResponse invocation_response = 5; + + // Host sends cancel message to attempt to cancel an invocation. + // If an invocation is cancelled, host will receive an invocation response with status cancelled. + InvocationCancel invocation_cancel = 21; + + // Worker logs a message back to the host + RpcLog rpc_log = 2; + + FunctionEnvironmentReloadRequest function_environment_reload_request = 25; + + FunctionEnvironmentReloadResponse function_environment_reload_response = 26; + } +} + +// Process.Start required info +// connection details +// protocol type +// protocol version + +// Worker sends the host information identifying itself +message StartStream { + // id of the worker + string worker_id = 2; +} + +// Host requests the worker to initialize itself +message WorkerInitRequest { + // version of the host sending init request + string host_version = 1; + + // A map of host supported features/capabilities + map capabilities = 2; + + // inform worker of supported categories and their levels + // i.e. Worker = Verbose, Function.MyFunc = None + map log_categories = 3; +} + +// Worker responds with the result of initializing itself +message WorkerInitResponse { + // Version of worker + string worker_version = 1; + // A map of worker supported features/capabilities + map capabilities = 2; + + // Status of the response + StatusResult result = 3; +} + +// Used by the host to determine success/failure/cancellation +message StatusResult { + // Indicates Failure/Success/Cancelled + enum Status { + Failure = 0; + Success = 1; + Cancelled = 2; + } + // Status for the given result + Status status = 4; + + // Specific message about the result + string result = 1; + + // Exception message (if exists) for the status + RpcException exception = 2; + + // Captured logs or relevant details can use the logs property + repeated RpcLog logs = 3; +} + +// TODO: investigate grpc heartbeat - don't limit to grpc implemention + +// Message is empty by design - Will add more fields in future if needed +message WorkerHeartbeat {} + +// Warning before killing the process after grace_period +// Worker self terminates ..no response on this +message WorkerTerminate { + google.protobuf.Duration grace_period = 1; +} + +// Host notifies worker of file content change +message FileChangeEventRequest { + // Types of File change operations (See link for more info: https://msdn.microsoft.com/en-us/library/t6xf43e0(v=vs.110).aspx) + enum Type { + Unknown = 0; + Created = 1; + Deleted = 2; + Changed = 4; + Renamed = 8; + All = 15; + } + + // type for this event + Type type = 1; + + // full file path for the file change notification + string full_path = 2; + + // Name of the function affected + string name = 3; +} + +// Indicates whether worker reloaded successfully or needs a restart +message WorkerActionResponse { + // indicates whether a restart is needed, or reload succesfully + enum Action { + Restart = 0; + Reload = 1; + } + + // action for this response + Action action = 1; + + // text reason for the response + string reason = 2; +} + +// NOT USED +message WorkerStatusRequest{ +} + +// NOT USED +message WorkerStatusResponse { +} + +message FunctionEnvironmentReloadRequest { + // Environment variables from the current process + map environment_variables = 1; + // Current directory of function app + string function_app_directory = 2; +} + +message FunctionEnvironmentReloadResponse { + // Status of the response + StatusResult result = 3; +} + +// Host tells the worker to load a Function +message FunctionLoadRequest { + // unique function identifier (avoid name collisions, facilitate reload case) + string function_id = 1; + + // Metadata for the request + RpcFunctionMetadata metadata = 2; + + // A flag indicating if managed dependency is enabled or not + bool managed_dependency_enabled = 3; +} + +// Worker tells host result of reload +message FunctionLoadResponse { + // unique function identifier + string function_id = 1; + + // Result of load operation + StatusResult result = 2; + // TODO: return type expected? + + // Result of load operation + bool is_dependency_downloaded = 3; +} + +// Information on how a Function should be loaded and its bindings +message RpcFunctionMetadata { + // TODO: do we want the host's name - the language worker might do a better job of assignment than the host + string name = 4; + + // base directory for the Function + string directory = 1; + + // Script file specified + string script_file = 2; + + // Entry point specified + string entry_point = 3; + + // Bindings info + map bindings = 6; + + // Is set to true for proxy + bool is_proxy = 7; +} + +// Host requests worker to invoke a Function +message InvocationRequest { + // Unique id for each invocation + string invocation_id = 1; + + // Unique id for each Function + string function_id = 2; + + // Input bindings (include trigger) + repeated ParameterBinding input_data = 3; + + // binding metadata from trigger + map trigger_metadata = 4; + + // Populates activityId, tracestate and tags from host + RpcTraceContext trace_context = 5; +} + +// Host sends ActivityId, traceStateString and Tags from host +message RpcTraceContext { + // This corresponds to Activity.Current?.Id + string trace_parent = 1; + + // This corresponds to Activity.Current?.TraceStateString + string trace_state = 2; + + // This corresponds to Activity.Current?.Tags + map attributes = 3; +} + +// Host requests worker to cancel invocation +message InvocationCancel { + // Unique id for invocation + string invocation_id = 2; + + // Time period before force shutdown + google.protobuf.Duration grace_period = 1; // could also use absolute time +} + +// Worker responds with status of Invocation +message InvocationResponse { + // Unique id for invocation + string invocation_id = 1; + + // Output binding data + repeated ParameterBinding output_data = 2; + + // data returned from Function (for $return and triggers with return support) + TypedData return_value = 4; + + // Status of the invocation (success/failure/canceled) + StatusResult result = 3; +} + +// Used to encapsulate data which could be a variety of types +message TypedData { + oneof data { + string string = 1; + string json = 2; + bytes bytes = 3; + bytes stream = 4; + RpcHttp http = 5; + sint64 int = 6; + double double = 7; + CollectionBytes collection_bytes = 8; + CollectionString collection_string = 9; + CollectionDouble collection_double = 10; + CollectionSInt64 collection_sint64 = 11; + } +} + +// Used to encapsulate collection string +message CollectionString { + repeated string string = 1; +} + +// Used to encapsulate collection bytes +message CollectionBytes { + repeated bytes bytes = 1; +} + +// Used to encapsulate collection double +message CollectionDouble { + repeated double double = 1; +} + +// Used to encapsulate collection sint64 +message CollectionSInt64 { + repeated sint64 sint64 = 1; +} + +// Used to describe a given binding on invocation +message ParameterBinding { + // Name for the binding + string name = 1; + + // Data for the binding + TypedData data = 2; +} + +// Used to describe a given binding on load +message BindingInfo { + // Indicates whether it is an input or output binding (or a fancy inout binding) + enum Direction { + in = 0; + out = 1; + inout = 2; + } + + // Indicates the type of the data for the binding + enum DataType { + undefined = 0; + string = 1; + binary = 2; + stream = 3; + } + + // Type of binding (e.g. HttpTrigger) + string type = 2; + + // Direction of the given binding + Direction direction = 3; + + DataType data_type = 4; +} + +// Used to send logs back to the Host +message RpcLog { + // Matching ILogger semantics + // https://github.com/aspnet/Logging/blob/9506ccc3f3491488fe88010ef8b9eb64594abf95/src/Microsoft.Extensions.Logging/Logger.cs + // Level for the Log + enum Level { + Trace = 0; + Debug = 1; + Information = 2; + Warning = 3; + Error = 4; + Critical = 5; + None = 6; + } + + // Category of the log. Defaults to User if not specified. + enum RpcLogCategory { + User = 0; + System = 1; + } + + // Unique id for invocation (if exists) + string invocation_id = 1; + + // TOD: This should be an enum + // Category for the log (startup, load, invocation, etc.) + string category = 2; + + // Level for the given log message + Level level = 3; + + // Message for the given log + string message = 4; + + // Id for the even associated with this log (if exists) + string event_id = 5; + + // Exception (if exists) + RpcException exception = 6; + + // json serialized property bag, or could use a type scheme like map + string properties = 7; + + // Category of the log. Either user(default) or system. + RpcLogCategory log_category = 8; +} + +// Encapsulates an Exception +message RpcException { + // Source of the exception + string source = 3; + + // Stack trace for the exception + string stack_trace = 1; + + // Textual message describing the exception + string message = 2; +} + +// Http cookie type. Note that only name and value are used for Http requests +message RpcHttpCookie { + // Enum that lets servers require that a cookie shouldn't be sent with cross-site requests + enum SameSite { + None = 0; + Lax = 1; + Strict = 2; + ExplicitNone = 3; + } + + // Cookie name + string name = 1; + + // Cookie value + string value = 2; + + // Specifies allowed hosts to receive the cookie + NullableString domain = 3; + + // Specifies URL path that must exist in the requested URL + NullableString path = 4; + + // Sets the cookie to expire at a specific date instead of when the client closes. + // It is generally recommended that you use "Max-Age" over "Expires". + NullableTimestamp expires = 5; + + // Sets the cookie to only be sent with an encrypted request + NullableBool secure = 6; + + // Sets the cookie to be inaccessible to JavaScript's Document.cookie API + NullableBool http_only = 7; + + // Allows servers to assert that a cookie ought not to be sent along with cross-site requests + SameSite same_site = 8; + + // Number of seconds until the cookie expires. A zero or negative number will expire the cookie immediately. + NullableDouble max_age = 9; +} + +// TODO - solidify this or remove it +message RpcHttp { + string method = 1; + string url = 2; + map headers = 3; + TypedData body = 4; + map params = 10; + string status_code = 12; + map query = 15; + bool enable_content_negotiation= 16; + TypedData rawBody = 17; + repeated RpcClaimsIdentity identities = 18; + repeated RpcHttpCookie cookies = 19; +} diff --git a/protobuf/src/proto/identity/ClaimsIdentityRpc.proto b/protobuf/src/proto/identity/ClaimsIdentityRpc.proto new file mode 100644 --- /dev/null +++ b/protobuf/src/proto/identity/ClaimsIdentityRpc.proto @@ -0,0 +1,26 @@ +syntax = "proto3"; +// protobuf vscode extension: https://marketplace.visualstudio.com/items?itemName=zxh404.vscode-proto3 + +option java_package = "com.microsoft.azure.functions.rpc.messages"; + +import "shared/NullableTypes.proto"; + +// Light-weight representation of a .NET System.Security.Claims.ClaimsIdentity object. +// This is the same serialization as found in EasyAuth, and needs to be kept in sync with +// its ClaimsIdentitySlim definition, as seen in the WebJobs extension: +// https://github.com/Azure/azure-webjobs-sdk-extensions/blob/dev/src/WebJobs.Extensions.Http/ClaimsIdentitySlim.cs +message RpcClaimsIdentity { + NullableString authentication_type = 1; + NullableString name_claim_type = 2; + NullableString role_claim_type = 3; + repeated RpcClaim claims = 4; +} + +// Light-weight representation of a .NET System.Security.Claims.Claim object. +// This is the same serialization as found in EasyAuth, and needs to be kept in sync with +// its ClaimSlim definition, as seen in the WebJobs extension: +// https://github.com/Azure/azure-webjobs-sdk-extensions/blob/dev/src/WebJobs.Extensions.Http/ClaimSlim.cs +message RpcClaim { + string value = 1; + string type = 2; +} diff --git a/protobuf/src/proto/shared/NullableTypes.proto b/protobuf/src/proto/shared/NullableTypes.proto new file mode 100644 --- /dev/null +++ b/protobuf/src/proto/shared/NullableTypes.proto @@ -0,0 +1,30 @@ +syntax = "proto3"; +// protobuf vscode extension: https://marketplace.visualstudio.com/items?itemName=zxh404.vscode-proto3 + +option java_package = "com.microsoft.azure.functions.rpc.messages"; + +import "google/protobuf/timestamp.proto"; + +message NullableString { + oneof string { + string value = 1; + } +} + +message NullableDouble { + oneof double { + double value = 1; + } +} + +message NullableBool { + oneof bool { + bool value = 1; + } +} + +message NullableTimestamp { + oneof timestamp { + google.protobuf.Timestamp value = 1; + } +} diff --git a/src/Azure/Functions/Bindings/Blob.hs b/src/Azure/Functions/Bindings/Blob.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Bindings/Blob.hs @@ -0,0 +1,106 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +module Azure.Functions.Bindings.Blob +( ConnectionName(..) +, BlobBinding(..) +, ReceivedBlob(..) +, Blob(..) +) +where + +import Azure.Functions.Bindings.Class +import Azure.Functions.Bindings.Shared +import Azure.Functions.Internal.Lens (orError) +import Control.Arrow ((&&&)) +import Data.Aeson (FromJSON, ToJSON (..), Value (Null), decodeStrict', object, (.=)) +import Data.ByteString (ByteString) +import Data.Coerce (coerce) +import Data.Functor ((<&>)) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) +import qualified Data.Set as Set +import Data.Text (Text) +import qualified Data.Text.Encoding as Text +import GHC.Generics (Generic) +import Lens.Family (view, (&), (.~), (^.)) +import Lens.Family.Stock (at) +import Network.URI (URI, parseURI) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields + +data BlobBinding = BlobBinding + { blobBindingConnectionName :: ConnectionName + , blobBindingPathPattern :: Text + } deriving (Generic) + +data ReceivedBlob = ReceivedBlob + { receivedBlobContent :: ByteString + , receivedBlobName :: Text + , receivedBlobUri :: URI + , receivedBlobMetadata :: Map Text Text + , receivedBlobTriggerMetadata :: Map Text Text + , receivedBlobProperties :: Value + } deriving (Show, Generic) + +data Blob = Blob + { sentBlobContent :: ByteString + } deriving (Show, Generic) + +instance ToInBinding BlobBinding where + toInBindingJSON v = object + [ "type" .= ("blobTrigger" :: Text) + , "direction" .= ("in" :: Text) + , "name" .= ("blobData" :: Text) + , "path" .= blobBindingPathPattern v + , "connection" .= coerce @_ @Text (blobBindingConnectionName v) + ] + +instance ToOutBinding BlobBinding where + toOutBindingJSON v = object + [ "type" .= ("blob" :: Text) + , "direction" .= ("out" :: Text) + , "name" .= ("$return" :: Text) + , "path" .= blobBindingPathPattern v + , "connection" .= coerce @_ @Text (blobBindingConnectionName v) + ] + +instance InMessage ReceivedBlob where + type InBinding ReceivedBlob = BlobBinding + fromInvocationRequest req = do + let idata = req ^. inputData <&> (view name &&& view data') & Map.fromList + let tmeta = req ^. triggerMetadata + + let orMissing fld = orError ("Unable to parse " <> fld) + + content <- idata ^. at "blobData" >>= getBytes & orMissing "blobData" + name <- tmeta ^. at "BlobTrigger" >>= getText & orMissing "BlobTrigger" + uri <- tmeta ^. at "Uri" >>= decodeJson >>= parseURI & orMissing "Uri" + props <- tmeta ^. at "Properties" >>= decodeJson & maybe (pure Null) pure + bmeta <- tmeta ^. at "Metadata" >>= decodeJson & maybe (pure mempty) pure + + pure ReceivedBlob + { receivedBlobContent = content + , receivedBlobName = name + , receivedBlobUri = uri + , receivedBlobMetadata = bmeta + , receivedBlobTriggerMetadata = Map.withoutKeys tmeta (Set.fromList ["Properties", "sys", "Metadata", "BlobTrigger"]) & Map.mapMaybe getText + , receivedBlobProperties = props + } + +instance OutMessage Blob where + type OutBinding Blob = BlobBinding + toInvocationResponse resp = + let + td = defMessage @TypedData + & maybe'data' .~ Just (TypedData'Bytes (sentBlobContent resp)) + + in defMessage @InvocationResponse + & returnValue .~ td + & result .~ (defMessage & status .~ StatusResult'Success) + diff --git a/src/Azure/Functions/Bindings/Class.hs b/src/Azure/Functions/Bindings/Class.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Bindings/Class.hs @@ -0,0 +1,40 @@ +-- {-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +module Azure.Functions.Bindings.Class +where + +import Data.Aeson (ToJSON, Value (Null)) +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) +import Data.Proxy (Proxy) +import Data.Text (Text) +import GHC.Generics (Generic) +import Lens.Family ((&), (.~)) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields + +class ToInBinding ctx where + toInBindingJSON :: ctx -> Value + +class ToOutBinding ctx where + toOutBindingJSON :: ctx -> Value + +class (ToInBinding (InBinding a)) => InMessage a where + type InBinding a :: * + fromInvocationRequest :: InvocationRequest -> Either Text a + +class (ToOutBinding (OutBinding a)) => OutMessage a where + type OutBinding a + toInvocationResponse :: a -> InvocationResponse + +instance OutMessage () where + type OutBinding () = () + toInvocationResponse _ = + defMessage @InvocationResponse + & result .~ (defMessage & status .~ StatusResult'Success) + +instance ToOutBinding () where + toOutBindingJSON _ = Null + diff --git a/src/Azure/Functions/Bindings/HTTP.hs b/src/Azure/Functions/Bindings/HTTP.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Bindings/HTTP.hs @@ -0,0 +1,106 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +module Azure.Functions.Bindings.HTTP +( HttpRequest(..) +, HttpResponse(..) +, HttpBinding(..) +, module Azure.Functions.Bindings.Class +) +where + +import Azure.Functions.Bindings.Class +import Azure.Functions.Internal.Lens (toEither) +import Data.Aeson ((.=)) +import qualified Data.Aeson as Aeson +import Data.ByteString (ByteString) +import qualified Data.Map as Map +import Data.Map.Strict (Map) +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.Encoding as Text +import GHC.Generics (Generic) +import Lens.Family (LensLike, Phantom, to, view, (&), (.~), (^.)) +import Lens.Family.Stock (at) +import Network.URI (URI, parseURI) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields + +data HttpBinding = HttpBinding + +data HttpRequest = HttpRequest + { httpRequestMethod :: Text + , httpRequestUrl :: URI + , httpRequestQuery :: Map Text Text + , httpRequestHeaders :: Map Text Text + , httpRequestBody :: Maybe ByteString + } deriving (Show, Eq, Generic) + +data HttpResponse = HttpResponse + { httpResponseStatus :: Int + , httpResponseBody :: Maybe ByteString + , httpResponseHeaders :: Map Text Text + } deriving (Show, Eq, Generic) + +instance InMessage HttpRequest where + type InBinding HttpRequest = HttpBinding + fromInvocationRequest req = + req ^. triggerMetadata . at "$request" . toEither "Unable to find $request parameter" + >>= view (maybe'http . toEither "Unexpected payload, RpcHttp is expected") + >>= fromRpcHttp + +instance OutMessage HttpResponse where + type OutBinding HttpResponse = HttpBinding + toInvocationResponse resp = + let + td = defMessage @TypedData + & maybe'data' .~ fmap TypedData'Bytes (httpResponseBody resp) + + ht = defMessage @RpcHttp + & headers .~ httpResponseHeaders resp + & body .~ td + + in defMessage @InvocationResponse + & returnValue .~ (defMessage & http .~ ht) + & result .~ (defMessage & status .~ StatusResult'Success) + +instance ToInBinding HttpBinding where + toInBindingJSON _ = Aeson.object + [ "type" .= ("httpTrigger" :: Text) + , "direction" .= ("in" :: Text) + , "name" .= ("req" :: Text) + ] + +instance ToOutBinding HttpBinding where + toOutBindingJSON _ = Aeson.object + [ "type" .= ("http" :: Text) + , "direction" .= ("out" :: Text) + , "name" .= ("$return" :: Text) + ] + +fromRpcHttp :: RpcHttp -> Either Text HttpRequest +fromRpcHttp req = do + uri <- req ^. url . to (Text.unpack) . to parseURI . toEither "Unable to parse URI" + + pure HttpRequest + { httpRequestMethod = req ^. method + , httpRequestUrl = uri + , httpRequestQuery = req ^. query + , httpRequestHeaders = req ^. headers + , httpRequestBody = (req ^. maybe'rawBody >>= fromTypedData) + } + +fromTypedData :: TypedData -> Maybe ByteString +fromTypedData td = + td ^. maybe'data' >>= \case + TypedData'String v -> Just (Text.encodeUtf8 v) + TypedData'Json v -> Just (Text.encodeUtf8 v) + TypedData'Bytes v -> Just v + TypedData'Stream v -> Just v + _ -> Nothing + diff --git a/src/Azure/Functions/Bindings/ServiceBus.hs b/src/Azure/Functions/Bindings/ServiceBus.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Bindings/ServiceBus.hs @@ -0,0 +1,118 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +module Azure.Functions.Bindings.ServiceBus +( QueueName(..) +, ConnectionName(..) +, ServiceBusBinding(..) +, ReceivedMessage(..) +) +where + +import Azure.Functions.Bindings.Class +import Azure.Functions.Bindings.Shared +import Azure.Functions.Internal.Lens (orError) +import Control.Applicative (Alternative, (<|>)) +import Control.Arrow ((&&&)) +import Data.Aeson (FromJSON, ToJSON (..), Value (Null), decodeStrict', object, (.=)) +import Data.ByteString (ByteString) +import Data.Coerce (coerce) +import Data.Functor ((<&>)) +import Data.Int (Int32, Int64) +import qualified Data.List as List +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Maybe (fromMaybe) +import Data.String (IsString) +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.Encoding as Text +import Data.Time (UTCTime) +import GHC.Generics (Generic) +import Lens.Family (view, (&), (^.)) +import Lens.Family.Stock (at) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields +import Text.Read (readMaybe) + +deliveryCountKey = "DeliveryCount" +deadLetterSourceKey = "DeadLetterSource" +expirationTimeKey = "ExpiresAtUtc" +enqueuedTimeKey = "EnqueuedTimeUtc" +messageIdKey = "MessageId" +contentTypeKey = "ContentType" +replyToKey = "ReplyTo" +sequenceNumberKey = "SequenceNumber" +toKey = "To" +labelKey = "Label" +correlationIdKey = "CorrelationId" +userPropertiesKey = "UserProperties" + +newtype QueueName = QueueName Text deriving (Show, Eq, IsString, Generic) + +data ServiceBusBinding = ServiceBusBinding + { serviceBusConnectionName :: ConnectionName + , serviceBusQueueName :: QueueName + } + +instance ToInBinding ServiceBusBinding where + toInBindingJSON v = object + [ "name" .= ("queueTrigger" :: Text) + , "type" .= ("serviceBusTrigger" :: Text) + , "direction" .= ("in" :: Text) + , "queueName" .= coerce @_ @Text (serviceBusQueueName v) + , "connection" .= coerce @_ @Text (serviceBusConnectionName v) + , "accessRights" .= ("Listen" :: Text) + ] + +data ReceivedMessage = ReceivedMessage + { receivedMessageId :: Text -- ^ The user-defined value that Service Bus can use to identify duplicate messages, if enabled. + , receivedMessageBody :: ByteString -- ^ The message that triggered the function. + , receivedMessageDeliveryCount :: Int32 -- ^ The number of deliveries. + , receivedMessageDeadLetterSource :: Maybe Text -- ^ The dead letter source. + , receivedMessageEnqueuedTime :: UTCTime -- ^ The time that the message was enqueued. + , receivedMessageExpirationTime :: UTCTime -- ^ The time that the message expires. + , receivedMessageContentType :: Maybe Text -- ^ The content type identifier utilized by the sender and receiver for application specific logic. + , receivedMessageReplyTo :: Maybe Text -- ^ The reply to queue address. + , receivedMessageSequenceNumber :: Int64 -- ^ The unique number assigned to a message by the Service Bus. + , receivedMessageTo :: Maybe Text -- ^ The send to address. + , receivedMessageLabel :: Maybe Text -- ^ The application specific label. + , receivedMessageCorrelationId :: Maybe Text -- ^ The correlation ID. + , receivedMessageUserProperties :: Map Text Text -- ^ The application specific message properties. + } deriving (Show, Eq, Generic) + +instance InMessage ReceivedMessage where + type InBinding ReceivedMessage = ServiceBusBinding + fromInvocationRequest req = do + let idata = req ^. inputData <&> (view name &&& view data') & Map.fromList + let metadata = req ^. triggerMetadata + + let orMissing fld = orError ("Unable to parse " <> fld) + + mid <- metadata ^. at messageIdKey >>= view maybe'string & orMissing messageIdKey + seq <- metadata ^. at sequenceNumberKey >>= decodeJson & orMissing sequenceNumberKey + body <- idata ^. at "queueTrigger" >>= getText & orMissing "queueTrigger" + delCnt <- metadata ^. at deliveryCountKey >>= decodeJson & orMissing deliveryCountKey + queued <- metadata ^. at enqueuedTimeKey >>= decodeJson & orMissing enqueuedTimeKey + expire <- metadata ^. at expirationTimeKey >>= decodeJson & orMissing expirationTimeKey + + pure ReceivedMessage + { receivedMessageId = mid + , receivedMessageBody = Text.encodeUtf8 body + , receivedMessageDeliveryCount = delCnt + , receivedMessageDeadLetterSource = metadata ^. at deadLetterSourceKey >>= getText + , receivedMessageExpirationTime = expire + , receivedMessageEnqueuedTime = queued + , receivedMessageContentType = metadata ^. at contentTypeKey >>= getText + , receivedMessageReplyTo = metadata ^. at replyToKey >>= getText + , receivedMessageSequenceNumber = seq + , receivedMessageTo = metadata ^. at toKey >>= getText + , receivedMessageLabel = metadata ^. at labelKey >>= getText + , receivedMessageCorrelationId = metadata ^. at correlationIdKey >>= getText + , receivedMessageUserProperties = metadata ^. at userPropertiesKey >>= decodeJson & fromMaybe mempty + } diff --git a/src/Azure/Functions/Bindings/Shared.hs b/src/Azure/Functions/Bindings/Shared.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Bindings/Shared.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +module Azure.Functions.Bindings.Shared +where + +import Control.Applicative (Alternative, (<|>)) +import Data.Aeson (FromJSON, decodeStrict') +import Data.ByteString (ByteString) +import Data.String (IsString) +import Data.Text (Text) +import qualified Data.Text.Encoding as Text +import GHC.Generics (Generic) +import Lens.Family ((^.)) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields + +newtype ConnectionName = ConnectionName Text deriving (Show, Eq, IsString, Generic) + +decodeJson :: FromJSON a => TypedData -> Maybe a +decodeJson d = + d ^. maybe'json >>= decodeStrict' . Text.encodeUtf8 +{-# INLINE decodeJson #-} + +getText :: TypedData -> Maybe Text +getText d = + d ^. maybe'data' >>= \case + TypedData'String v -> Just v + TypedData'Json v -> Just v + _ -> Nothing + +getBytes :: TypedData -> Maybe ByteString +getBytes td = + td ^. maybe'data' >>= \case + TypedData'Bytes v -> Just v + TypedData'Stream v -> Just v + _ -> Nothing + +infixl 3 <||> +(<||>) :: Alternative f => (a -> f x) -> (a -> f x) -> a -> f x +f <||> g = \a -> f a <|> g a +{-# INLINE (<||>) #-} diff --git a/src/Azure/Functions/Commands/Init.hs b/src/Azure/Functions/Commands/Init.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Commands/Init.hs @@ -0,0 +1,83 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +module Azure.Functions.Commands.Init +where + +import qualified Azure.Functions.Internal.Templates as Tpl +import Azure.Functions.Registry (RegisteredFunction (..), Registry (..)) +import Control.Monad (forM_, unless) +import Data.Aeson (Value (Null), object, (.=)) +import qualified Data.Aeson as Aeson +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import qualified Data.Text as Text +import qualified Data.Text.IO as Text +import GHC.Generics (Generic) +import Options.Applicative +import System.Directory as Dir +import System.Environment (getExecutablePath) +import System.FilePath (takeFileName, (<.>), ()) + +data Options = Options + { scriptRoot :: String + , syncExtensions :: Bool + } deriving (Generic) + +optionsParser :: Parser Options +optionsParser = Options + <$> strOption + ( long "script-root" + <> metavar "DIR" + <> help "The script root directory to initialize the application in" + ) + <*> switch + ( long "sync-extensions" + <> help "Synchronize the Azure Function binding extensions" + ) + +initCommand :: Registry -> Parser (IO ()) +initCommand registry = runInitCommand registry <$> optionsParser + +runInitCommand :: Registry -> Options -> IO () +runInitCommand registry opts = do + funcRoot <- Dir.getCurrentDirectory + execPath <- getExecutablePath + projectRoot <- Dir.makeAbsolute (scriptRoot opts) + + let execFileName = takeFileName execPath + + let workerDir = projectRoot "workers" "haskell" + let workerExecPath = workerDir execFileName + + createDirectoryIfMissing True workerDir + + Dir.copyFileWithMetadata "host.json" (projectRoot "host.json") + Dir.copyFileWithMetadata "local.settings.json" (projectRoot "local.settings.json") + + Dir.copyFileWithMetadata execPath workerExecPath + + Tpl.writeFile (workerDir "worker.config.json") Tpl.workerConfigJson [("execPath", Text.pack workerExecPath)] + + forM_ (Map.toList (registeredFunctions registry)) (uncurry $ initRegisteredFunction projectRoot) + +initRegisteredFunction :: FilePath -> Text -> RegisteredFunction -> IO () +initRegisteredFunction path functionName function = do + let functionPath = path (Text.unpack functionName) + let jsonFile = functionPath "function.json" + let functionFile = functionPath (Text.unpack functionName) <.> "hs" + + createDirectoryIfMissing True functionPath + + Aeson.encodeFile jsonFile $ + object [ "generatedBy" .= ("azure-functions-haskell-worker" :: Text) + , "disabled" .= False + , "bindings" .= filter (/= Null) + [ registeredInBinding function + , registeredOutBinding function + ] + ] + + Text.writeFile functionFile "" + + + diff --git a/src/Azure/Functions/Commands/Run.hs b/src/Azure/Functions/Commands/Run.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Commands/Run.hs @@ -0,0 +1,191 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TypeApplications #-} + +module Azure.Functions.Commands.Run +where + +import Azure.Functions.Internal.Runtime (Runtime, createRuntime, getRuntimeFunction, loadRuntimeFunction, runningFunction) +import qualified Azure.Functions.Internal.Templates as Tpl +import Azure.Functions.Registry (RegisteredFunction (..), Registry (..)) +import qualified Azure.Functions.Registry as Registry +import Control.Concurrent.Async (async, link) +import Control.Concurrent.Chan (newChan, readChan, writeChan) +import Control.Monad (forM_, mapM_, void) +import Control.Monad.Except (runExceptT) +import Control.Monad.IO.Class (liftIO) +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import qualified Data.Text as Text +import GHC.Generics (Generic) +import Lens.Family ((&), (.~), (^.)) +import Network.GRPC.HTTP2.ProtoLens (RPC (..)) +import Options.Applicative +import System.Directory as Dir +import System.Environment (getArgs, getExecutablePath) +import System.FilePath (()) + +import Network.GRPC.Client as GRPC +import Network.GRPC.Client.Helpers as GRPC +import Network.HTTP2.Client (HostName, PortNumber, TooMuchConcurrency) +import Network.HTTP2.Client.Exceptions (ClientIO) + +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) + +import Proto.FunctionRpc +import qualified Proto.FunctionRpc_Fields as Fields +import Proto.FunctionRpc_Helpers (failureStatus, successStatus, toResponse) + +import Data.Version (showVersion) +import Paths_azure_functions_worker (version) + +import Azure.Functions.Bindings.Class (fromInvocationRequest, toInvocationResponse) +import Azure.Functions.Bindings.HTTP (HttpRequest (..), HttpResponse (..)) +import Azure.Functions.Bindings.ServiceBus (ReceivedMessage (..)) + +type RequestId = Text +type WorkerId = Text + +data Options = Options + { host :: HostName + , port :: PortNumber + , workerId :: WorkerId + , requestId :: RequestId + } deriving (Show, Generic) + +optionsParser :: Parser Options +optionsParser = Options + <$> strOption + ( long "host" + <> metavar "HOST" + <> help "Run on the specified host" + ) + <*> option auto + ( long "port" + <> metavar "PORT" + <> help "Run gRPC channel on the specified port" + ) + <*> strOption + ( long "workerId" + <> help "Worker ID" + ) + <*> strOption + ( long "requestId" + <> help "Request ID" + ) + <* option (auto @Int) + ( long "grpcMaxMessageLength" + <> help "The maximum message length to use for gRPC messages" + ) + +runCommand :: Registry -> Parser (IO ()) +runCommand registry = runRunCommand registry <$> optionsParser + +data WorkerState = Unitialised | Started + +runRunCommand :: Registry -> Options -> IO () +runRunCommand registry opts = do + runtime <- createRuntime + outgoingMessageChannel <- newChan + + let cfg = GRPC.grpcClientConfigSimple (host opts) (port opts) False + + let startMsg = defMessage @StartStream & Fields.workerId .~ (workerId opts) + let initMsg = defMessage @StreamingMessage + & Fields.requestId .~ (requestId opts) + & Fields.maybe'content .~ Just (StreamingMessage'StartStream startMsg) + + writeChan outgoingMessageChannel initMsg + + let rpc = RPC :: RPC FunctionRpc "eventStream" + void $ runExceptT $ do + client <- GRPC.setupGrpcClient cfg + + notTooMuch $ rawGeneralStream rpc client + outgoingMessageChannel -- incoming messages loop state, handler needs to write to the channel + (handleIncoming runtime) -- incoming messages loop + outgoingMessageChannel -- outgoing messages loop state, reads messages from the channel and sends the, + handleOutgoing -- outgoing messages loop + + print "-----------------------------------------------------------------------------" + + where + handleIncoming runtime chan = \case + RecvMessage msg -> do + liftIO $ runAsync $ do + out <- handleEnvelope registry runtime msg + forM_ out (writeChan chan) + pure chan + _ -> pure chan + + handleOutgoing chan = do + msg <- liftIO (readChan chan) + pure (chan, SendMessage Uncompressed msg) + + runAsync f = async f >>= link + +handleEnvelope :: Registry -> Runtime -> StreamingMessage -> IO [StreamingMessage] +handleEnvelope registry runtime req = do + appendFile "/tmp/msg" ("Request:\n============================================================\n") + appendFile "/tmp/msg" (show req <> "\n\n") + + let rid = req ^. Fields.requestId + resp <- case req ^. Fields.maybe'content of + Nothing -> pure [] + Just c -> case c of + StreamingMessage'WorkerInitRequest msg -> sequence [toResponse req <$> handleWorkerInit msg] + StreamingMessage'FunctionLoadRequest msg -> sequence [toResponse req <$> handleFunctionLoad registry runtime msg] + StreamingMessage'InvocationRequest msg -> sequence [toResponse req <$> handleInvocation runtime msg] + msg -> pure [] + + appendFile "/tmp/msg" ("Response:\n------------------------------------------------------------\n") + appendFile "/tmp/msg" (show resp <> "\n\n") + pure resp + +handleWorkerInit :: WorkerInitRequest -> IO WorkerInitResponse +handleWorkerInit msg = do + let resp = defMessage @WorkerInitResponse + & Fields.workerVersion .~ Text.pack (showVersion version) + & Fields.result .~ successStatus ("Haskell worker loaded, version: " <> Text.pack (showVersion version)) + + pure resp + +handleFunctionLoad :: Registry -> Runtime -> FunctionLoadRequest -> IO FunctionLoadResponse +handleFunctionLoad registry runtime req = do + let funcName = req ^. Fields.metadata . Fields.name + let funcId = req ^. Fields.functionId + + let resp = defMessage @FunctionLoadResponse + & Fields.functionId .~ (req ^. Fields.functionId) + + case Registry.getFunction registry funcName of + Nothing -> + pure $ resp & Fields.result .~ failureStatus ("Unable to find function: " <> funcName) + Just func -> do + loadRuntimeFunction runtime funcId func + pure $ resp & Fields.result .~ successStatus ("Loaded function: " <> funcName) + +handleInvocation :: Runtime -> InvocationRequest -> IO InvocationResponse +handleInvocation runtime req = do + let funcId = req ^. Fields.functionId + mbFunction <- getRuntimeFunction runtime funcId + case mbFunction of + Nothing -> pure $ + defMessage @InvocationResponse + & Fields.invocationId .~ (req ^. Fields.invocationId) + & Fields.result .~ failureStatus ("Unable to find running function with ID: " <> funcId) + Just rfunc -> do + resp <- runningFunction rfunc req + pure $ resp & Fields.invocationId .~ (req ^. Fields.invocationId) + +------------------------------------------------------------------------------- + +notTooMuch :: ClientIO (Either TooMuchConcurrency a) -> ClientIO a +notTooMuch f = do + ma <- f + case ma of + Left _ -> error "Too much concurrency. TODO: do something smarter" + Right a -> pure a diff --git a/src/Azure/Functions/Contract.hs b/src/Azure/Functions/Contract.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Contract.hs @@ -0,0 +1,7 @@ +module Azure.Functions.Contract +( module X +) +where + +import Proto.FunctionRpc as X +import Proto.FunctionRpc_Fields as X diff --git a/src/Azure/Functions/Function.hs b/src/Azure/Functions/Function.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Function.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE StrictData #-} +module Azure.Functions.Function +where + +import Azure.Functions.Bindings.Class +import Data.Aeson (Value) +import Data.Functor ((<&>)) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.ProtoLens.Runtime.Data.ProtoLens as PL +import Data.Text (Text) +import GHC.Generics (Generic) +import Lens.Family ((&), (.~), (^.)) +import Proto.FunctionRpc +import qualified Proto.FunctionRpc_Fields as Fields +import Proto.FunctionRpc_Helpers (failureStatus, rpcLogError, rpcLogInfo, toResponse, toResponseLogError') + +data Function env i o = Function + { inBinding :: InBinding i + , outBinding :: OutBinding o + , initEnv :: IO env + , func :: env -> i -> IO o + } diff --git a/src/Azure/Functions/Internal/Lens.hs b/src/Azure/Functions/Internal/Lens.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Internal/Lens.hs @@ -0,0 +1,12 @@ +module Azure.Functions.Internal.Lens +where + +import Lens.Family + +toEither :: Phantom f => a -> LensLike f (Maybe b) t (Either a b) b +toEither = to . orError +{-# INLINE toEither #-} + +orError :: a -> Maybe b -> Either a b +orError msg = maybe (Left msg) Right +{-# INLINE orError #-} diff --git a/src/Azure/Functions/Internal/Runtime.hs b/src/Azure/Functions/Internal/Runtime.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Internal/Runtime.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} +module Azure.Functions.Internal.Runtime +where + +import Azure.Functions.Registry (RegisteredFunction (..)) +import Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO, readTVarIO) +import Control.Monad.STM (atomically) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import GHC.Generics (Generic) +import Proto.FunctionRpc + +type FunctionId = Text + +data RunningFunction = RunningFunction + { runningFunction :: InvocationRequest -> IO InvocationResponse + } deriving (Generic) + +data Runtime = Runtime + { runtimeFunctions :: TVar (Map Text RunningFunction) + } + +createRuntime :: IO Runtime +createRuntime = Runtime <$> newTVarIO mempty + +loadRuntimeFunction :: Runtime -> FunctionId -> RegisteredFunction -> IO () +loadRuntimeFunction runtime fid RegisteredFunction{registeredEnvFactory, registeredFunction} = do + func <- registeredFunction <$> registeredEnvFactory + atomically $ do + modifyTVar' (runtimeFunctions runtime) $ Map.insert fid (RunningFunction func) + +getRuntimeFunction :: Runtime -> FunctionId -> IO (Maybe RunningFunction) +getRuntimeFunction runtime fid = do + Map.lookup fid <$> readTVarIO (runtimeFunctions runtime) diff --git a/src/Azure/Functions/Internal/Templates.hs b/src/Azure/Functions/Internal/Templates.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Internal/Templates.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} +module Azure.Functions.Internal.Templates +where + +import Control.Monad (unless) +import Data.Text (Text) +import qualified Data.Text.IO as Text +import Prelude hiding (writeFile) +import System.Directory (createDirectoryIfMissing, doesFileExist) +import System.FilePath (takeDirectory) +import Text.Glabrous (Template) +import qualified Text.Glabrous as Tpl +import Text.RawString.QQ + + +workerConfigJson :: Template +workerConfigJson = toTemplate "worker.config.json" [r| +{ + "description": { + "arguments": [ + "run" + ], + "defaultExecutablePath": "{{execPath}}", + "extensions": [ + ".hs" + ], + "language": "haskell" + } +} +|] + +writeFile :: FilePath -> Template -> [(Text, Text)] -> IO () +writeFile file tpl vars = do + createDirectoryIfMissing True (takeDirectory file) + let content = Tpl.process tpl (Tpl.fromList vars) + Text.writeFile file content + +writeFileIfNotExist :: FilePath -> Template -> [(Text, Text)] -> IO () +writeFileIfNotExist file tpl vars = do + ok <- doesFileExist file + unless ok $ writeFile file tpl vars + +toTemplate :: String -> Text -> Template +toTemplate name tpl = + case Tpl.fromText tpl of + Left err -> error $ "Unable to create template: " <> name <> ". Error: " <> err + Right tpl' -> tpl' diff --git a/src/Azure/Functions/Registry.hs b/src/Azure/Functions/Registry.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Registry.hs @@ -0,0 +1,61 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE NamedFieldPuns #-} +module Azure.Functions.Registry +where + +import Azure.Functions.Function +import Azure.Functions.Bindings.Class +import Data.Aeson (Value) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import GHC.Generics (Generic) +import Lens.Family ((&), (.~), (^.)) +import Proto.FunctionRpc +import qualified Proto.FunctionRpc_Fields as Fields +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) +import Proto.FunctionRpc_Helpers (failureStatus) + +data RegisteredFunction = forall env. RegisteredFunction + { registeredInBinding :: Value + , registeredOutBinding :: Value + , registeredEnvFactory :: IO env + , registeredFunction :: env -> InvocationRequest -> IO InvocationResponse + } + +newtype Registry = Registry + { registeredFunctions :: Map Text RegisteredFunction + } deriving (Generic, Semigroup) + deriving Monoid via (Map Text RegisteredFunction) + +getFunction :: Registry -> Text -> Maybe RegisteredFunction +getFunction registry name = + Map.lookup name (registeredFunctions registry) + +register :: (InMessage i, OutMessage o) + => Text + -> Function env i o + -> Registry +register functionName function = + Registry $ Map.singleton functionName RegisteredFunction + { registeredInBinding = toInBindingJSON (inBinding function) + , registeredOutBinding = toOutBindingJSON (outBinding function) + , registeredEnvFactory = initEnv function + , registeredFunction = invoke + } + where + invoke ctx req = do + case fromInvocationRequest req of + Left err -> pure $ + defMessage + & Fields.invocationId .~ (req ^. Fields.invocationId) + & Fields.result .~ failureStatus ("Unable to parse request: " <> err) + Right req' -> do + resp <- toInvocationResponse <$> (func function ctx) req' + pure $ resp & Fields.invocationId .~ (req ^. Fields.invocationId) + diff --git a/src/Azure/Functions/Worker.hs b/src/Azure/Functions/Worker.hs new file mode 100644 --- /dev/null +++ b/src/Azure/Functions/Worker.hs @@ -0,0 +1,25 @@ +module Azure.Functions.Worker +( runWorker +) +where + +import Azure.Functions.Commands.Init (initCommand) +import Azure.Functions.Commands.Run (runCommand) +import Azure.Functions.Registry (Registry) +import Control.Monad (join) +import Options.Applicative +import System.Environment (getArgs) +import qualified System.IO as IO + +runWorker :: Registry -> IO () +runWorker registry = do + IO.hSetBuffering IO.stderr IO.LineBuffering + join $ customExecParser + (prefs $ showHelpOnEmpty <> showHelpOnError) + (info (commands registry <**> helper) idm) + +commands :: Registry -> Parser (IO ()) +commands registry = hsubparser + ( command "init" (info (initCommand registry) idm) + <> command "run" (info (runCommand registry) idm) + ) diff --git a/src/Proto/FunctionRpc_Helpers.hs b/src/Proto/FunctionRpc_Helpers.hs new file mode 100644 --- /dev/null +++ b/src/Proto/FunctionRpc_Helpers.hs @@ -0,0 +1,102 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TypeApplications #-} +module Proto.FunctionRpc_Helpers +where + +import Data.ProtoLens.Runtime.Data.ProtoLens (defMessage) +import Data.Text (Text) +import Lens.Family ((&), (.~), (^.)) +import Proto.FunctionRpc +import Proto.FunctionRpc_Fields + +class ToStreamingMessageContent a where + toStreamingMessageContent :: a -> StreamingMessage'Content + +instance ToStreamingMessageContent WorkerInitResponse where + toStreamingMessageContent = StreamingMessage'WorkerInitResponse + +instance ToStreamingMessageContent FunctionLoadResponse where + toStreamingMessageContent = StreamingMessage'FunctionLoadResponse + +instance ToStreamingMessageContent InvocationResponse where + toStreamingMessageContent = StreamingMessage'InvocationResponse + +instance ToStreamingMessageContent FunctionEnvironmentReloadResponse where + toStreamingMessageContent = StreamingMessage'FunctionEnvironmentReloadResponse + +instance ToStreamingMessageContent WorkerStatusResponse where + toStreamingMessageContent = StreamingMessage'WorkerStatusResponse + +instance ToStreamingMessageContent WorkerActionResponse where + toStreamingMessageContent = StreamingMessage'WorkerActionResponse + +instance ToStreamingMessageContent WorkerHeartbeat where + toStreamingMessageContent = StreamingMessage'WorkerHeartbeat + +instance ToStreamingMessageContent RpcLog where + toStreamingMessageContent = StreamingMessage'RpcLog + +toResponse :: ToStreamingMessageContent a + => StreamingMessage + -> a + -> StreamingMessage +toResponse req resp = + defMessage + & requestId .~ (req ^. requestId) + & maybe'content .~ Just (toStreamingMessageContent resp) + +toResponseLogError :: ToStreamingMessageContent a + => StreamingMessage + -> Either Text a + -> StreamingMessage +toResponseLogError req resp = toResponseLogError' req id resp + +toResponseLogError' :: ToStreamingMessageContent a + => StreamingMessage + -> (RpcLog -> RpcLog) + -> Either Text a + -> StreamingMessage +toResponseLogError' req updateLog resp = + let content = either (toStreamingMessageContent . updateLog . rpcLogError) toStreamingMessageContent resp + in defMessage + & requestId .~ (req ^. requestId) + & maybe'content .~ Just content + +rpcLogMessage :: RpcLog'Level -> Text -> RpcLog +rpcLogMessage lvl msg = + defMessage + & level .~ lvl + & message .~ msg + +rpcLogInfo :: Text -> RpcLog +rpcLogInfo = rpcLogMessage RpcLog'Information + +rpcLogError :: Text -> RpcLog +rpcLogError = rpcLogMessage RpcLog'Error + +successStatus :: Text -> StatusResult +successStatus msg = + defMessage @StatusResult + & status .~ StatusResult'Success + & result .~ msg + & logs .~ [ + rpcLogInfo msg + ] + & exception .~ ( + defMessage @RpcException + & message .~ msg + ) + +failureStatus :: Text -> StatusResult +failureStatus msg = + defMessage @StatusResult + & status .~ StatusResult'Failure + & result .~ msg + & logs .~ [ + rpcLogError msg + ] + & exception .~ ( + defMessage @RpcException + & message .~ msg + ) diff --git a/test/MyLibTest.hs b/test/MyLibTest.hs new file mode 100644 --- /dev/null +++ b/test/MyLibTest.hs @@ -0,0 +1,6 @@ +module Main where + +main :: IO () +main = do + putStrLn "Hello, Haskell!" +