more filesHEADmaster

1 files changed, 88 insertions, 0 deletions

diff --git a/src/orchestration/vmb_consumer.py b/src/orchestration/vmb_consumer.py
new file mode 100644
index 0000000..3d6ebe6
--- /dev/null
+++ b/src/orchestration/vmb_consumer.py
@@ -0,0 +1,88 @@
+import os
+import pprint
+
+from django.conf import settings
+from pulsar import Client, AuthenticationTLS
+from .vmb_messages import *
+
+class VMBConsumer(object):
+
+    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'src.far_edge_ops_api.settings')
+
+    TLS_CERT_FILE = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_CERT_FILE"])
+    TLS_KEY_FILE = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_KEY_FILE"])
+    TLS_TRUST_CERTS_FILE_PATH = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_TRUST_CERTS_FILE_PATH"])
+
+    # Each environment(Rocklin, Branchburg, AWS Non Prod/PLE/STG/PROD has its own service url
+    SERVICE_URL = settings.VMB['SERVICE_URL']
+    #SERVICE_URL = "pulsar://localhost:6650/"
+
+    auth = AuthenticationTLS(TLS_CERT_FILE,TLS_KEY_FILE)
+
+    def get_topic_message(self, topic, topic_schema):
+
+        client = Client(self.SERVICE_URL, tls_trust_certs_file_path=self.TLS_TRUST_CERTS_FILE_PATH,
+                        tls_allow_insecure_connection=False, authentication=self.auth)
+        #client = Client(SERVICE_URL)
+
+        topic = settings.VMB["VMB_TOPICS"][topic]
+        consumer = client.subscribe(topic=topic, subscription_name='sub-2', schema=JsonSchema(eval(topic_schema)))
+        pp = pprint.PrettyPrinter(indent=4)
+        try:
+            while True:
+                msg = consumer.receive()
+                ex = msg.value()
+                try:
+                    print("Received message {}".format(ex))
+                    consumer.acknowledge(msg)
+                except:
+                    # Message failed to be processed
+                    consumer.negative_acknowledge(msg)
+        except KeyboardInterrupt:
+            pass
+
+        client.close()
+
+import sys, getopt
+
+def main(argv):
+    try:
+        opts, args = getopt.getopt(argv,"ht:",["topic="])
+    except getopt.GetoptError:
+        print('vmb_consumer.py -t <topic>')
+        sys.exit(2)
+    for opt, arg in opts:
+        if opt == '-h':
+            print('vmb_consumer.py -t [cluster_status|namespace|kubeconfig|image_status>')
+            sys.exit()
+        elif opt in ("-t", "--topic"):
+            topic = arg
+            print('topic is ' + topic)
+
+    vmbConsumer = VMBConsumer()
+
+
+    if (topic == 'namespace'):
+        # 1. Namespace Creation
+        print('subscribe to namespace topic ...')
+        vmbConsumer.get_topic_message(topic='TOPIC_NAMESPACE_CREATION', topic_schema='NameSpaceMessage')
+
+    elif(topic == 'cluster_status'):
+        # 2. Cluster Status
+        print('subscribe to cluster_status topic ...')
+        vmbConsumer.get_topic_message(topic='TOPIC_CLUSTER_STATUS', topic_schema='ClusterStatusMessage')
+
+    elif(topic == 'kubeconfig'):
+        # 3. Kubeconfig Token
+        print('subscribe to kubeconfig topic ...')
+        vmbConsumer.get_topic_message(topic='TOPIC_KUBECONFIG_TOKEN', topic_schema='KubeconfigMessage')
+
+    elif(topic == 'image_status'):
+        # 4. Images Status
+        print('subscribe to image_status topic ...')
+        vmbConsumer.get_topic_message(topic='TOPIC_IMAGE_STATUS', topic_schema='ImagesStatusMessage')
+    else:
+        print('unsupport topic: ' + topic)
+
+if __name__ == "__main__":
+    main(sys.argv[1:])
\ No newline at end of file