moving configs

sample/config.edn

@@ -1,21 +1,21 @@
-#:pg{:users #:users{:usr/replication #:usr{:password "secret"}}
-     :hba [["local" "all" "all" "trust"]
-           ["local" "replication" "replication" "trust"]
-           ["host"  "replication" "replication" "0.0.0.0/0" "md5"]
-           ["host"  "all" "all" "0.0.0.0/0" "md5"]]
-     :failover {:tx/timeout 5000}
-     :health   {:tx/timeout 3000
-                :chk/master {:simple-read "SELECT 1"
-                             :snapshot "SELECT txid_current_snapshot()"
-                             :transaction-id "SELECT txid_current()"}
-                :chk/replica {:simple-read "SELECT 1"
-                              :snapshot "SELECT txid_current_snapshot()"}}
-     :postgres #:pg{:max_connections 100
-                    :listen_addresses "*"
-                    :wal_level "logical"
-                    :hot_standby "on"
-                    :archive_mode "on"
-                    :archive_timeout 60
-                    :max_wal_senders 10
-                    :wal_keep_segments 100
-                    :shared_buffers  "128MB"}}
+#:glb{:users #:usrs{:usr/replication #:usr{:password "secret"}}
+      :hba [["local" "all" "all" "trust"]
+            ["local" "replication" "replication" "trust"]
+            ["host"  "replication" "replication" "0.0.0.0/0" "md5"]
+            ["host"  "all" "all" "0.0.0.0/0" "md5"]]
+      :failover {:tx/timeout 5000}
+      :health   {:tx/timeout 3000
+                 :chk/master {:simple-read "SELECT 1"
+                              :snapshot "SELECT txid_current_snapshot()"
+                              :transaction-id "SELECT txid_current()"}
+                 :chk/replica {:simple-read "SELECT 1"
+                               :snapshot "SELECT txid_current_snapshot()"}}
+      :postgres #:pg{:max_connections 100
+                     :listen_addresses "*"
+                     :wal_level "logical"
+                     :hot_standby "on"
+                     :archive_mode "on"
+                     :archive_timeout 60
+                     :max_wal_senders 10
+                     :wal_keep_segments 100
+                     :shared_buffers  "128MB"}}

sample/node-1.edn

@@ -1,9 +1,10 @@
-#:cfg{:name "node-1"
+#:lcl{:name "node-1"
+      :host "127.0.0.1"
       :postgres-bin "/usr/lib/postgresql/9.5/bin"
       :data-dir "/tmp/node-1"
       :web #:web{:port 8080}
-      :atomix   #:ax{:port 4444
-                     :host "127.0.0.1"}
+      :atomix   #:ax{:port 4444}
       :postgres #:pg{:port 5434
-                     :host "127.0.0.1"
+                     :restore_command "cp /tmp/wallogs/pg_xlog/%f %p"
+                     :archive_cleanup_command "rm -f %r"
                      :archive_command "cp %p /tmp/wallogs/%p"}}

sample/node-2.edn

@@ -1,11 +1,10 @@
-#:cfg{:name "node-2"
+#:lcl{:name "node-2"
       :postgres-bin "/usr/lib/postgresql/9.5/bin"
       :data-dir "/tmp/node-2"
       :web #:web{:port 8081}
+      :host "127.0.0.1"
       :atomix   #:ax{:port 4445
-                     :host "127.0.0.1"
                      :join [{:host "127.0.0.1" :port 4444}]
                      :discovery ["http://localhost:8080/discovery"]}
       :postgres #:pg{:port 5435
-                     :host "127.0.0.1"
                      :archive_command "cp %p /tmp/wallogs/%p"}}

sample/node-3.edn

@@ -1,11 +1,10 @@
-#:cfg{:name "node-3"
+#:lcl{:name "node-3"
       :postgres-bin "/usr/lib/postgresql/9.5/bin"
       :data-dir "/tmp/node-3"
       :web #:web{:port 8082}
+      :host "127.0.0.1"
       :atomix   #:ax{:port 4446
-                     :host "127.0.0.1"
                      :join [{:host "127.0.0.1" :port 4444}]
                      :discovery ["http://localhost:8080/discovery"]}
       :postgres #:pg{:port 5435
-                     :host "127.0.0.1"
                      :archive_command "cp %p /tmp/wallogs/%p"}}

src/ankusha/atomix/core.clj

@@ -34,7 +34,7 @@
 (s/fdef replica :args :cfg/local-config)
 
 (defn- replica ^AtomixReplica
-  [{data-dir :cfg/data-dir {port :ax/port host :ax/host} :cfg/atomix :as cfg}]
+  [{data-dir :cfg/data-dir {port :ax/port} :cfg/atomix host :cfg/host :as cfg}]
   (log/info "Init replica at " (str  host ":" port))
   (sh/sh "mkdir" "-p" data-dir)
   (-> (AtomixReplica/builder (addr {:host host :port port}))

src/ankusha/config.clj

@@ -9,49 +9,48 @@
 (defn ipv4? [x] (re-matches #"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$" x))
 
 (s/def :web/port number?)
-(s/def :ax/host ipv4?)
-(s/def :pg/host ipv4?)
+(s/def :lcl/host ipv4?)
 
-(s/def :cfg/postgres-bin file-exists?)
-(s/def :cfg/data-dir file-path?)
-(s/def :cfg/web (s/keys :req [:web/port]))
-(s/def :cfg/atomix (s/keys :req [:ax/port :ax/host]))
-(s/def :cfg/postgres (s/keys :req [:pg/port :pg/host] :opt [:pg/archive_command]))
+(s/def :lcl/postgres-bin file-exists?)
+(s/def :lcl/data-dir file-path?)
+(s/def :lcl/web (s/keys :req [:web/port]))
+(s/def :lcl/atomix (s/keys :req [:ax/port]))
+(s/def :lcl/postgres  (s/keys :req [:pg/port :pg/archive_command]))
+(s/def :glb/postgres  (s/keys :opt [:pg/archive_command :pg/port]))
 
 
-(s/def :cfg/local-config (s/keys :req [:cfg/data-dir :cfg/name :cfg/postgres-bin :cfg/postgres]))
+(s/def :lcl/config (s/keys :req [:lcl/data-dir :lcl/name :lcl/postgres-bin :lcl/postgres :lcl/host]))
+(s/def :glb/config (s/keys :req [:glb/hba :glb/users :glb/postgres]))
 
 (defn load-config [filepath]
   (edn/read-string (slurp filepath)))
 
 (defn load-global [filepath]
-  (edn/read-string (slurp filepath)))
+  (let [cfg (load-config filepath)]
+    (if (s/valid? :glb/config cfg)
+      (do (state/assoc-in [:global-config] cfg) cfg)
+      (throw (Exception. (with-out-str (s/explain :glb/config cfg)))))))
 
 (defn load-local [filepath]
   (let [cfg (load-config filepath)]
-    (if (s/valid? :cfg/local-config cfg)
-      (do
-        (state/assoc-in [:local-config] cfg)
-        cfg)
-      (throw (Exception. (with-out-str (s/explain :cfg/local-config cfg)))))))
+    (if (s/valid? :lcl/config cfg)
+      (do (state/assoc-in [:local-config] cfg) cfg)
+      (throw (Exception. (with-out-str (s/explain :lcl/config cfg)))))))
 
 (defn local []
   (state/get-in [:local-config]))
 
+(defn global []
+  (state/get-in [:global-config]))
+
 
 (defn get-local [])
 
 (comment
 
-  (with-out-str
-    (s/explain ::local-config
-               (load-config "sample/node-1.edn")))
-
-  (load-config "sample/node-1.edn")
-
   (load-local "sample/node-1.edn")
 
-  (local)
+  (load-global "sample/config.edn")
 
 
 

src/ankusha/core.clj

@@ -26,8 +26,7 @@
 
 (defn parse-addrs [addrs]
   (map (fn [x]
-         (let [a (str/split x #":")]
-           {:host (get a 0) :port (Integer/parseInt (get a 1))}))
+         (let [a (str/split x #":")] {:host (get a 0) :port (Integer/parseInt (get a 1))}))
        (str/split addrs #",")))
 
 (defn bootstrap-replica [local-config-path addr-str]

src/ankusha/pg/config.clj

@@ -1,10 +1,11 @@
 (ns ankusha.pg.config
   (:require [clojure.string :as str]
             [clojure.tools.logging :as log]
+            [ankusha.config :as conf]
             [clojure.core.async :as async :refer [<!]]))
 
 (defn pg-data-dir [cfg & args]
-  (apply str (:data-dir cfg) "/pg" (when-not (empty? args) (str "/" (str/join "/" args)))))
+  (apply str (:lcl/data-dir cfg) "/pg" (when-not (empty? args) (str "/" (str/join "/" args)))))
 
 (defn to-config [cfg]
   (with-out-str
@@ -14,50 +15,39 @@
 (defn to-props [cfg]
   (str/join " " (for [[k v] cfg] (str (name k) "=" v))))
 
-(to-props {:a 1 :b 3})
-
 (defn mk-config [cfg]
   (to-config (:config cfg)))
 
-(defn mk-hba [cfg]
-  (with-out-str
-    (doseq [ks (:hba cfg)]
-      (println (str/join "\t" (map name ks))))))
+(defn hba [gcfg lcfg]
+  (->> gcfg
+       :glb/hba
+       (map #(str/join "\t" %))
+       (str/join "\n")))
+
+(defn config [gcfg lcfg]
+  (to-config (merge (:glb/postgres gcfg)
+                    (:lcl/postgres lcfg)
+                    {:unix_socket_directories (pg-data-dir lcfg) 
+                     :log_line_prefix (str "<" (:lcl/name lcfg) ">")})))
 
-(defn mk-recovery [master-cfg replica-cfg]
+(defn mk-recovery [master-cfg lcfg]
   (to-config
    {:standby_mode "on"
     :primary_conninfo (to-props {:host (:host master-cfg) 
                                  :port (:port master-cfg)
                                  :user (get-in master-cfg [:user :name]) 
                                  :password (get-in master-cfg [:user :password])})
-    :restore_command  (str "cp /tmp/wallogs/pg_xlog/%f %p")
-    :archive_cleanup_command "rm -f %r"}))
-
-(defn update-hba [cfg]
-  (let [hba (mk-hba cfg)
-        conf_path (pg-data-dir cfg "pg_hba.conf")]
-    (log/info "Update pg_hbal:\n" hba)
-    (spit conf_path hba)))
-
-(defn update-config [cfg]
-  (let [pgconf (mk-config cfg)
-        conf_path (pg-data-dir cfg "postgresql.conf")]
-    (log/info "Update config:\n" pgconf)
-    (spit conf_path pgconf)))
-
-(defn update-recovery [master-cfg replica-cfg]
-  (let [txt (mk-recovery master-cfg replica-cfg)
-        conf_path (pg-data-dir replica-cfg "/recovery.conf")]
-    (log/info "Update recovery.conf [" conf_path "]\n" txt )
-    (spit conf_path txt)))
-
-;; (defn sighup-params [cfg]
-;;   (pg/exec (conn-uri cfg)
-;;            {:select [:name]
-;;             :from [:pg_settings]
-;;             :where [:= :context "sighup"]}))
+    :restore_command (:pg/restore_command lcfg) 
+    :archive_cleanup_command (:pg/archive_cleanup_command lcfg)}))
 
 (comment
-  (update-config config)
-  (reconfigure config))
+  (conf/load-local "sample/node-1.edn")
+  (conf/load-global "sample/config.edn")
+
+  (println "--------")
+  (println (hba (conf/global) (conf/local)))
+
+  (println "--------")
+  (println (config (conf/global) (conf/local)))
+
+  )

src/ankusha/pg/core.clj

@@ -75,7 +75,7 @@
       (throw (Exception. (str "Unable to connect to postgres"))))))
 
 (defn initdb [cfg]
-  (log/info "[" (:name cfg) "] " "initdb -D" (pg-data-dir cfg))
+  (log/info "[" (:cfg/name cfg) "] " "initdb -D" (pg-data-dir cfg))
   (let [res (sh/sh (bin "initdb") "-D" (pg-data-dir cfg))]
     (if (= 0 (:exit res))
       (log/info (:out res))
@@ -88,41 +88,37 @@
     (when (.exists (io/file pth))
       (first (str/split (slurp pth) #"\n")))))
 
-(defn start []
-  (if-let [cfg (get-node)]
-    (if (.exists (io/file (pg-data-dir cfg "/postmaster.pid")))
-      (log/info "Postgres already started")
-      (let [res (pg_ctl cfg :start)]
-        (state/assoc-in [:pg :process] (pid))
-        res))
-    (log/info "Node not initialized")))
+(defn start [lcfg]
+  (if (.exists (io/file (pg-data-dir lcfg "/postmaster.pid")))
+    (log/info "Postgres already started")
+    (let [res (pg_ctl lcfg :start)]
+      (state/assoc-in [:pg :process] (pid))
+      res)))
 
 (defn stop []
   (pg_ctl (get-node) :stop))
 
 
-(defn create-user [cfg {nm :name pswd :password}]
-  (log/info "Create user" nm )
-  (psql cfg (str  "CREATE USER " nm " WITH SUPERUSER PASSWORD '" pswd "'")))
+(defn create-user [lcfg users]
+  (doseq [[nm usr] users]
+    (log/info "Create user" nm)
+    (psql lcfg (str  "CREATE USER " nm " WITH SUPERUSER PASSWORD '" (:usr/password usr) "'"))))
 
-(defn master [opts]
-  (let [cfg (merge (node-config opts)
-                   {:user {:name "ankus" :password (str (java.util.UUID/randomUUID))}})]
-    (initdb cfg)
-    (pg-config/update-config cfg)
-    (pg-config/update-hba cfg)
-    (state/assoc-in [:pg] cfg)
-    (start)
-    (wait-pg cfg 10 "SELECT 1")
-    (create-user cfg (:user cfg))
-    cfg))
+
+(defn master [gcfg lcfg]
+  (initdb lcfg)
+  (pg-config/config gcfg lcfg)
+  (pg-config/hba gcfg lcfg)
+  (start lcfg)
+  (wait-pg lcfg 10 "SELECT 1")
+  (create-user lcfg (:glb/users gcfg)))
 
 (defn kill [pid sig]
   (sh/sh "kill" (str "-" (str/upper-case (name sig))) pid))
 
-(defn replica [parent-cfg replica-opts]
-  (let [cfg (node-config replica-opts)
-        pgpass-path (str (:data-dir replica-opts) "/.pgpass")
+(defn replica [parent-cfg {data-dir :cfg/data-dir :as replica-cfg}]
+  (let [cfg (node-config replica-cfg)
+        pgpass-path (str data-dir "/.pgpass")
         args [(bin "pg_basebackup")
               "-w"
               "-h" (:host parent-cfg)
@@ -132,8 +128,8 @@
               "-D" (pg-data-dir cfg)
               :env {"PGPASSFILE" pgpass-path}]]
 
-    (sh! "mkdir" "-p" (:data-dir cfg))
-    (sh! "chmod" "0700" (:data-dir cfg))
+    (sh! "mkdir" "-p" data-dir)
+    (sh! "chmod" "0700" data-dir)
     (spit pgpass-path
           (let [{{u :name pwd :password} :user h :host p :port} parent-cfg]
             (str h ":" p ":*:" u ":" pwd "\n")))
@@ -166,44 +162,14 @@
   (sh/sh "rm" "-rf" "/tmp/wallogs")
   (sh/sh "mkdir" "-p" "/tmp/wallogs/pg_xlog")
 
-  (clean-up)
-
-  (for [x (enumeration-seq (java.net.NetworkInterface/getNetworkInterfaces))]
-    x)
-
-  (def cfg
-    {:name "node-1"
-     :host "127.0.0.1"
-     :port 5434
-     :data-dir "/tmp/node-1"})
-
-  (def master-cfg (master cfg))
-
-  (:user master-cfg)
-  (:host master-cfg)
-
-  (start)
-  (stop)
-
-  (sh! "rm" "-rf" "/tmp/node-2")
+  (sh/sh "rm" "-rf" "/tmp/node-1/pg")
 
-  (state/with-node "node-2"
-    (replica master-cfg
-             {:name "node-2"
-              :host "127.0.0.1"
-              :port 5435
-              :data-dir "/tmp/node-2"}))
+  (require '[ankusha.config :as conf])
 
-  (start)
-  (state/with-node "node-2" (stop))
+  (conf/load-local "sample/node-1.edn")
 
+  (conf/load-global "sample/config.edn")
 
-  (state/with-node "node-3"
-    (replica master-cfg 
-             {:name "node-3"
-              :host "127.0.0.1"
-              :port 5436
-              :data-dir "/tmp/node-3"}))
+  (master (conf/global) (conf/local))
 
-  (state/with-node "node-3" (stop))
   )