initial

Author: Oleg Noga

.gitignore

@@ -1,14 +1,5 @@
-pom.xml
-pom.xml.asc
-*.jar
-*.class
-/lib/
-/classes/
-/target/
-/checkouts/
-.lein-deps-sum
-.lein-repl-history
-.lein-plugins/
-.lein-failures
-.nrepl-port
-.cpcache/
+/target
+/.nrepl-port
+/.idea
+/*.iml
+/*.lein-*

README.adoc

@@ -0,0 +1,38 @@
+= Consul agent client for clojure
+
+Any clojure service can register and send heartbeats to consul agent.
+
+== Start continuous heartbeat
+
+Service will be registered initially.
+If consul or third party deregisters service, registration will be renewed"
+
+[source,clojure]
+----
+(start-heartbeat {:id                                "my-service" ; service id
+                  :address                           "127.0.0.1"  ; service remote address
+                  :port                              8081
+                  :ttl                               "15s"        ; timeout since last heartbeat to set critical state
+                  :deregister-critical-service-after "1m"         ; timeout for consul to deregister service in critical state
+                  :interval-ms                       7500})       ; heartbeat interval, milliseconds
+----
+
+== Stop continuous heartbeat
+
+[source,clojure]
+----
+(stop-heartbeat "my-service")
+----
+
+== Single hearbeat
+
+Service will be registered if not registered yet.
+
+[source,clojure]
+----
+(heartbeat {:id                                "my-service"
+            :address                           "127.0.0.1"
+            :port                              8081
+            :ttl                               "15s"
+            :deregister-critical-service-after "1m"})
+----

project.clj

@@ -0,0 +1,11 @@
+(defproject consul-clj "0.1.0"
+  :description "Clojure library for registering service with consul agent"
+  :url "https://github.com/tbt-post/consul-clj"
+  :license {:name "MIT"}
+  :dependencies [[http-kit "2.3.0"]
+                 [cheshire "5.10.0"]
+                 [org.clojure/core.async "1.3.610"]
+                 [org.clojure/tools.logging "1.1.0"]]
+  :profiles {:uberjar {:aot :all}
+             :dev     {:dependencies [[org.clojure/clojure "1.10.1"]]}}
+  :plugins [[lein-ancient "0.6.15"]])

src/consul/agent.clj

@@ -0,0 +1,101 @@
+(ns consul.agent
+  (:require [cheshire.core :as json]
+            [clojure.core.async :refer [go-loop timeout <!]]
+            [clojure.string :as string]
+            [clojure.tools.logging :as log]
+            [org.httpkit.client :as http]))
+
+
+(def ^:dynamic *consul-url* "http://127.0.0.1:8500/shit")
+(def ^:private alive (atom #{}))
+
+
+(defn- default-callback [{:keys [status] :as response}]
+  (when (or
+          (nil? status)
+          (>= status 400))
+    (log/warn "Error requesting consul agent: " response)))
+
+
+(defn- http-get [path]
+  (log/debug "GET" (str *consul-url* "/v1/agent" path))
+  (http/request {:url (str *consul-url* "/v1/agent" path)}))
+
+
+(defn- http-put [path body callback]
+  (log/debug "PUT" (str *consul-url* "/v1/agent" path) "\n" (json/encode body {:pretty true}))
+  (http/request {:method :put
+                 :url    (str *consul-url* "/v1/agent" path)
+                 :body   (if body (json/encode body) "")}
+                (or callback default-callback)))
+
+
+(defn service-list []
+  (http-get "/services"))
+
+
+(defn service-details [service-id]
+  (http-get (str "/service/" service-id)))
+
+
+(defn service-register [request & [callback]]
+  (http-put "/service/register" request callback))
+
+
+(defn service-deregister [service-id & [callback]]
+  (http-put (str "/service/deregister/" service-id) nil callback))
+
+
+(defn check-update [check-id status & [callback]]
+  {:pre [(#{:passing :warning :critical} status)]}
+  (http-put (str "/check/update/" check-id) {:status status} callback))
+
+
+(defn check-update-with-register
+  "check update with registration if not registered yet"
+  [check-id check-status register-request]
+  (let [consul-url *consul-url*]
+    (check-update check-id check-status
+                  (fn [{:keys [status body] :as response}]
+                    (log/debug "-->" status body)
+                    (if (and status
+                             (>= status 400)
+                             (string/includes? body (format "\"%s\"" check-id))
+                             (string/includes? body "TTL"))
+                      (binding [*consul-url* consul-url]
+                        (service-register register-request
+                                          (fn [{:keys [status] :as response}]
+                                            (if (= status 200)
+                                              (binding [*consul-url* consul-url]
+                                                (check-update check-id check-status))
+                                              (default-callback response)))))
+                      (default-callback response))))))
+
+
+(defn heartbeat
+  "single heartbeat"
+  [{:keys [id name address port ttl deregister-critical-service-after]}]
+  (let [check-id (str id ":ttl-check")]
+    (check-update-with-register check-id :passing
+                                {:id      id
+                                 :name    (or name id)
+                                 :address address
+                                 :port    port
+                                 :check   {:CheckId                        check-id
+                                           :TTL                            ttl
+                                           :DeregisterCriticalServiceAfter deregister-critical-service-after}})))
+
+
+(defn start-heartbeat [{:keys [id interval-ms] :as params}]
+  (service-deregister id)
+  (swap! alive #(conj % [*consul-url* id]))
+  (go-loop []
+    (when (@alive [*consul-url* id])
+      (heartbeat params)
+      (<! (timeout interval-ms))
+      (recur))))
+
+
+(defn stop-heartbeat [service-id]
+  (swap! alive #(disj % [*consul-url* service-id]))
+  (service-deregister service-id))

test/consul/agent_test.clj

@@ -0,0 +1,36 @@
+(ns consul.agent-test
+  (:require [clojure.test :refer :all]
+            [consul.agent :as agent]
+            [clojure.core.async :refer [timeout <!!]]))
+
+
+(deftest test-1-start-stop-heartbeat
+  (testing "start-stop-heartbeat"
+    (binding [agent/*consul-url* "http://127.0.0.1:8500"]
+
+      (agent/start-heartbeat {:id                                "my-service"
+                              :address                           "127.0.0.1"
+                              :port                              8080
+                              :ttl                               "4s"
+                              :deregister-critical-service-after "1m"
+                              :interval-ms                       2000})
+
+      (<!! (timeout 500))
+
+      (let [{:keys [status body]} @(agent/service-details "my-service")]
+        (println body)
+        (is (= status 200)))
+
+      (<!! (timeout 8500))
+
+      (let [{:keys [status body]} @(agent/service-details "my-service")]
+        (println body)
+        (is (= status 200)))
+
+      (agent/stop-heartbeat "my-service")
+
+      (<!! (timeout 500))
+
+      (let [{:keys [status body]} @(agent/service-details "my-service")]
+        (println body)
+        (is (= status 404))))))