keepalived_status_check.py からロジック部分を status_check.py に切り出した

elements/keepalived-status-check/pre-install.d/10-keepalived-status-check-install

@@ -7,7 +7,8 @@ set -eu
 set -o pipefail
 
 install -m 0755 -o root -g root $(dirname $0)/keepalived_status_check.py /usr/local/bin/keepalived_status_check.py
+install -m 0755 -o root -g root $(dirname $0)/status_check.py /usr/local/bin/status_check.py
 sed -i -e 's,%%{OS_USERNAME}%%,'${OS_USERNAME}',g' \
        -e 's,%%{OS_PASSWORD}%%,'${OS_PASSWORD}',g' \
        -e 's,%%{OS_AUTH_URL}%%,'${OS_AUTH_URL}',g' \
-       /usr/local/bin/keepalived_status_check.py
+       /usr/local/bin/status_check.py

elements/keepalived-status-check/pre-install.d/keepalived_status_check.py

@@ -1,61 +1,19 @@
 #!/opt/amphora-agent-venv/bin/python3
 
-import os
 import sys
 import select
-import json
-import urllib3
-from keystoneauth1 import identity
-from keystoneauth1 import session
-from neutronclient.v2_0 import client
+import status_check
 
 
 # skeleton config parameters
 pollPeriod = 0.2 # the number of seconds between polling for new messages
 maxAtOnce = 1024  # max nbr of messages that are processed within one batch
 
-# App logic global variables
-def getNeutronClient():
-  username='%%{OS_USERNAME}%%'
-  password='%%{OS_PASSWORD}%%'
-  project_name='admin'
-  project_domain_id='default'
-  user_domain_id='default'
-  auth_url='%%{OS_AUTH_URL}%%'
-  auth = identity.Password(auth_url=auth_url,
-                           username=username,
-                           password=password,
-                           project_name=project_name,
-                           project_domain_id=project_domain_id,
-                           user_domain_id=user_domain_id)
-  sess = session.Session(auth=auth)
-  return client.Client(session=sess)
-
-def retrivePortID():
-  interface_file = "/var/lib/octavia/plugged_interfaces"
-  if not os.path.exists(interface_file):
-    return None
-
-  with open(interface_file, "r") as f:
-    mac_address = f.readline().split(" ")[0]
-
-  http = urllib3.PoolManager()
-  response = http.request("GET", "http://169.254.169.254/openstack/latest/network_data.json")
-  network_data = json.loads(response.data.decode("utf-8"))
-
-  info = list(filter((lambda n: n["ethernet_mac_address"] == mac_address), network_data["links"]))
-  return info[0]["vif_id"]
-
 def onInit():
   pass
 
 def onReceive(msgs):
-  port_id = retrivePortID()
-  if port_id is None:
-    return False
-
-  neutron = getNeutronClient()
-  neutron.update_port(port_id, {"port":{"admin_state_up": True}})
+  return status_check.update_port()
 
 def onExit():
   pass

elements/keepalived-status-check/pre-install.d/status_check.py

@@ -0,0 +1,50 @@
+#!/opt/amphora-agent-venv/bin/python3
+
+import os
+import json
+import urllib3
+from keystoneauth1 import identity
+from keystoneauth1 import session
+from neutronclient.v2_0 import client
+
+def getNeutronClient():
+  username='%%{OS_USERNAME}%%'
+  password='%%{OS_PASSWORD}%%'
+  project_name='admin'
+  project_domain_id='default'
+  user_domain_id='default'
+  auth_url='%%{OS_AUTH_URL}%%'
+  auth = identity.Password(auth_url=auth_url,
+                           username=username,
+                           password=password,
+                           project_name=project_name,
+                           project_domain_id=project_domain_id,
+                           user_domain_id=user_domain_id)
+  sess = session.Session(auth=auth)
+  return client.Client(session=sess)
+
+def retrivePortID():
+  interface_file = "/var/lib/octavia/plugged_interfaces"
+  if not os.path.exists(interface_file):
+    return None
+
+  with open(interface_file, "r") as f:
+    mac_address = f.readline().split(" ")[0]
+
+  http = urllib3.PoolManager()
+  response = http.request("GET", "http://169.254.169.254/openstack/latest/network_data.json")
+  network_data = json.loads(response.data.decode("utf-8"))
+
+  info = list(filter((lambda n: n["ethernet_mac_address"] == mac_address), network_data["links"]))
+  return info[0]["vif_id"]
+
+def update_port():
+  port_id = retrivePortID()
+  if port_id is None:
+    return False
+
+  neutron = getNeutronClient()
+  neutron.update_port(port_id, {"port":{"admin_state_up": True}})
+
+if __name__ == '__main__':
+  update_port()