blob: 9ec0103e95d6ed31d6933fec9c978e03abeac336 [file] [log] [blame]
raghunath dudyala01482a82016-08-01 12:29:16 +05301import sys, getopt
Matteo Scandolobf14f882016-06-02 10:01:34 -07002import socket
3import requests
4import urllib2
5import json
6import msgpack
7import collections
8import time, thread, threading
9
10from flask import request, Request, jsonify
11from flask import Flask
12from flask import make_response
13app = Flask(__name__)
14
15projects_map = {}
16xos_tenant_info_map = {}
17xos_instances_info_map = {}
18
raghunath dudyala01482a82016-08-01 12:29:16 +053019#gethostname() -- return the current hostname
20#gethostbyname() -- map a hostname to its IP number
21hostname = None
22xos_ip = None
23kafka_ip = xos_ip
24xos_port = 9000
25
26print "----------------------------------------------------------------------"
27print "xos_ip:",xos_ip
28print "kfka_ip:",kafka_ip
29print "----------------------------------------------------------------------"
30
Matteo Scandolobf14f882016-06-02 10:01:34 -070031use_kafka = True
32
33if use_kafka:
34 import kafka
raghunath dudyala01482a82016-08-01 12:29:16 +053035 import kafka_broker
36 from oslo_utils import netutils
37 #from kafka import TopicPartition
Matteo Scandolobf14f882016-06-02 10:01:34 -070038else:
39 UDP_IP = "0.0.0.0"
40 UDP_PORT = 12346
41
42@app.route('/autoscaledata',methods=['GET'])
43def autoscaledata():
44 response = app.make_response(json.dumps(projects_map.values()))
45 response.mimetype="application/json"
46 return response
47
48def acquire_xos_monitoring_channel():
raghunath dudyala01482a82016-08-01 12:29:16 +053049 global xos_ip
50
51 url = "http://" + xos_ip + ":" + str(xos_port) +"/api/tenant/monitoring/monitoringchannel/"
52 print url
Matteo Scandolobf14f882016-06-02 10:01:34 -070053 admin_auth=("padmin@vicci.org", "letmein") # use your XOS username and password
54 monitoring_channels = requests.get(url, auth=admin_auth).json()
55 ceilometer_url = None
56 if not monitoring_channels:
raghunath dudyala01482a82016-08-01 12:29:16 +053057 print 'No monitoring channels for this user...'
Matteo Scandolobf14f882016-06-02 10:01:34 -070058 return None
59 else:
60 monitoring_channel = monitoring_channels[0]
61 while not monitoring_channel['ceilometer_url']:
raghunath dudyala01482a82016-08-01 12:29:16 +053062 print 'Waiting for monitoring channel create'
Matteo Scandolobf14f882016-06-02 10:01:34 -070063 sleep(0.5)
64 monitoring_channel = requests.get(url, auth=admin_auth).json()[0]
Matteo Scandolobf14f882016-06-02 10:01:34 -070065 while True:
raghunath dudyala01482a82016-08-01 12:29:16 +053066 print 'Waiting for ceilometer proxy URL %s is available' % monitoring_channel['ceilometer_url']
Matteo Scandolobf14f882016-06-02 10:01:34 -070067 try:
68 response = urllib2.urlopen(monitoring_channel['ceilometer_url'],timeout=1)
69 break
70 except urllib2.HTTPError, e:
raghunath dudyala01482a82016-08-01 12:29:16 +053071 print 'HTTP error %s' % e.reason
Matteo Scandolobf14f882016-06-02 10:01:34 -070072 break
73 except urllib2.URLError, e:
raghunath dudyala01482a82016-08-01 12:29:16 +053074 print 'URL error %s' % e.reason
Matteo Scandolobf14f882016-06-02 10:01:34 -070075 pass
76 return monitoring_channel
77
78def print_samples():
79 print ""
80 print ""
81 for project in projects_map.keys():
82 print "service=%s slice=%s, alarm_state=%s" % (projects_map[project]['service'], projects_map[project]['slice'] if projects_map[project]['slice'] else project, projects_map[project]['alarm'])
83 for resource in projects_map[project]['resources'].keys():
84 print "resource=%s" % (projects_map[project]['resources'][resource]['xos_instance_info']['instance_name'] if projects_map[project]['resources'][resource]['xos_instance_info'] else resource)
85 for i in projects_map[project]['resources'][resource]['queue']:
86 print " time=%s val=%s" % ( i['timestamp'],i['counter_volume'])
87
88def periodic_print():
89 print_samples()
90 #Print every 1minute
91 threading.Timer(20, periodic_print).start()
92
93
94CPU_UPPER_THRESHOLD = 80 #80%
95CPU_LOWER_THRESHOLD = 30 #30%
96CPU_THRESHOLD_REPEAT = 3
97INITIAL_STATE = 'normal_config'
98SCALE_UP_EVALUATION = 'scale_up_eval'
99SCALE_DOWN_EVALUATION = 'scale_down_eval'
100SCALE_UP_ALARM = 'scale_up'
101SCALE_DOWN_ALARM = 'scale_down'
102
103def loadAllXosTenantInfo():
raghunath dudyala01482a82016-08-01 12:29:16 +0530104 global xos_ip
105
106 print "Loading all XOS tenant info"
107 url = "http://" + xos_ip + ":" + str(xos_port) +"/xos/controllerslices/"
Matteo Scandolobf14f882016-06-02 10:01:34 -0700108 admin_auth=("padmin@vicci.org", "letmein") # use your XOS username and password
109 controller_slices = requests.get(url, auth=admin_auth).json()
110 for cslice in controller_slices:
111 slice = requests.get(cslice['slice'], auth=admin_auth).json()
112 slice_name = slice['humanReadableName']
113 if slice['service']:
114 service = requests.get(slice['service'], auth=admin_auth).json()
115 service_name = service['humanReadableName']
116 else:
117 service_name = None
118 xos_tenant_info_map[cslice['tenant_id']] = {'service':service_name, 'slice':slice_name}
119 print "SRIKANTH: Project: %s Service:%s Slice:%s" % (cslice['tenant_id'],service_name,slice_name)
120
121def loadAllXosInstanceInfo():
raghunath dudyala01482a82016-08-01 12:29:16 +0530122 global xos_ip
123
124 print "Loading all XOS instance info"
125 url = "http://" + xos_ip + ":" + str(xos_port) +"/xos/instances/"
Matteo Scandolobf14f882016-06-02 10:01:34 -0700126 admin_auth=("padmin@vicci.org", "letmein") # use your XOS username and password
127 xos_instances = requests.get(url, auth=admin_auth).json()
128 for instance in xos_instances:
129 xos_instances_info_map[instance['instance_uuid']] = {'instance_name':instance['instance_name']}
130
131def getXosTenantInfo(project):
132 xos_tenant_info = xos_tenant_info_map.get(project, None)
133 if xos_tenant_info:
134 return xos_tenant_info
135 else:
136 loadAllXosTenantInfo()
137 xos_tenant_info = xos_tenant_info_map.get(project, None)
138 if not xos_tenant_info:
raghunath dudyala01482a82016-08-01 12:29:16 +0530139 print "Project %s has no associated XOS slice" % project
Matteo Scandolobf14f882016-06-02 10:01:34 -0700140 return xos_tenant_info
141
142def getXosInstanceInfo(resource):
143 xos_instance_info = xos_instances_info_map.get(resource, None)
144 if xos_instance_info:
145 return xos_instance_info
146 else:
147 loadAllXosInstanceInfo()
148 xos_instance_info = xos_instances_info_map.get(resource, None)
149 if not xos_instance_info:
raghunath dudyala01482a82016-08-01 12:29:16 +0530150 print "Resource %s has no associated XOS instance" % project
Matteo Scandolobf14f882016-06-02 10:01:34 -0700151 return xos_instance_info
152
153def handle_adjust_scale(project, adjust):
raghunath dudyala01482a82016-08-01 12:29:16 +0530154 global xos_ip
155
Matteo Scandolobf14f882016-06-02 10:01:34 -0700156 if (adjust != 'up') and (adjust != 'down'):
157 print "SRIKANTH: Invalid adjust value %s " % adjust
158 return
159 current_instances = len(projects_map[project]['resources'].keys())
160 if (current_instances <=1 and adjust == 'down'):
raghunath dudyala01482a82016-08-01 12:29:16 +0530161 print "%s is running with already minimum instances and can not scale down further " % project
Matteo Scandolobf14f882016-06-02 10:01:34 -0700162 return
163 if (current_instances >=2 and adjust == 'up'):
raghunath dudyala01482a82016-08-01 12:29:16 +0530164 print "%s is running with already maximum instances and can not scale up further " % project
Matteo Scandolobf14f882016-06-02 10:01:34 -0700165 return
166 #xos_tenant = getXosTenantInfo(project)
167 xos_service = projects_map[project]['service']
168 xos_slice = projects_map[project]['slice']
169 if not xos_service or not xos_slice:
raghunath dudyala01482a82016-08-01 12:29:16 +0530170 print "Can not handle adjust_scale for Project %s because not associated with any service or slice" % project
Matteo Scandolobf14f882016-06-02 10:01:34 -0700171 return
raghunath dudyala01482a82016-08-01 12:29:16 +0530172 print "SCALE %s for Project %s, Slice=%s, Service=%s from current=%d to new=%d" % (adjust, project, xos_slice, xos_service, current_instances, current_instances+1 if (adjust=='up') else current_instances-1)
Matteo Scandolobf14f882016-06-02 10:01:34 -0700173 query_params = {'service':xos_service, 'slice_hint':xos_slice, 'scale':current_instances+1 if (adjust=='up') else current_instances-1}
raghunath dudyala01482a82016-08-01 12:29:16 +0530174 url = "http://" + xos_ip + ":" + str(xos_port) + "/xoslib/serviceadjustscale/"
Matteo Scandolobf14f882016-06-02 10:01:34 -0700175 admin_auth=("padmin@vicci.org", "letmein") # use your XOS username and password
176 response = requests.get(url, params=query_params, auth=admin_auth).json()
177 print "SRIKANTH: XOS adjust_scale response: %s" % response
178
179def periodic_cpu_threshold_evaluator():
180 for project in projects_map.keys():
181 aggregate_cpu_util = sum([resource['queue'][-1]['counter_volume'] \
182 for resource in projects_map[project]['resources'].values()]) \
183 /len(projects_map[project]['resources'].keys())
184
185 if (projects_map[project]['alarm'] == INITIAL_STATE or
186 projects_map[project]['alarm'] == SCALE_UP_ALARM or
187 projects_map[project]['alarm'] == SCALE_DOWN_ALARM):
188 if aggregate_cpu_util > CPU_UPPER_THRESHOLD:
189 projects_map[project]['uthreadshold_count'] = 1
190 projects_map[project]['alarm'] = SCALE_UP_EVALUATION
191 if projects_map[project]['uthreadshold_count'] >= CPU_THRESHOLD_REPEAT:
192 projects_map[project]['alarm'] = SCALE_UP_ALARM
193 handle_adjust_scale(project, 'up')
194 elif aggregate_cpu_util < CPU_LOWER_THRESHOLD:
195 projects_map[project]['lthreadshold_count'] = 1
196 projects_map[project]['alarm'] = SCALE_DOWN_EVALUATION
197 if projects_map[project]['lthreadshold_count'] >= CPU_THRESHOLD_REPEAT:
198 projects_map[project]['alarm'] = SCALE_DOWN_ALARM
199 handle_adjust_scale(project, 'down')
200 else:
201 projects_map[project]['uthreadshold_count'] = 0
202 projects_map[project]['lthreadshold_count'] = 0
203 projects_map[project]['alarm'] = INITIAL_STATE
204 elif projects_map[project]['alarm'] == SCALE_UP_EVALUATION:
205 if aggregate_cpu_util > CPU_UPPER_THRESHOLD:
206 projects_map[project]['uthreadshold_count'] += 1
207 if projects_map[project]['uthreadshold_count'] >= CPU_THRESHOLD_REPEAT:
208 projects_map[project]['alarm'] = SCALE_UP_ALARM
209 handle_adjust_scale(project, 'up')
210 elif aggregate_cpu_util < CPU_LOWER_THRESHOLD:
211 projects_map[project]['lthreadshold_count'] += 1
212 projects_map[project]['alarm'] = SCALE_DOWN_EVALUATION
213 else:
214 projects_map[project]['uthreadshold_count'] = 0
215 projects_map[project]['alarm'] = INITIAL_STATE
216 elif projects_map[project]['alarm'] == SCALE_DOWN_EVALUATION:
217 if aggregate_cpu_util < CPU_LOWER_THRESHOLD:
218 projects_map[project]['lthreadshold_count'] += 1
219 if projects_map[project]['lthreadshold_count'] >= CPU_THRESHOLD_REPEAT:
220 projects_map[project]['alarm'] = SCALE_DOWN_ALARM
221 handle_adjust_scale(project, 'down')
222 elif aggregate_cpu_util > CPU_UPPER_THRESHOLD:
223 projects_map[project]['uthreadshold_count'] += 1
224 projects_map[project]['alarm'] = SCALE_UP_EVALUATION
225 else:
226 projects_map[project]['lthreadshold_count'] = 0
227 projects_map[project]['alarm'] = INITIAL_STATE
228 threading.Timer(20, periodic_cpu_threshold_evaluator).start()
229
raghunath dudyala01482a82016-08-01 12:29:16 +0530230def read_notification_from_ceilometer_over_kafka(parse_target):
231 print("Kafka target:",parse_target)
232 try :
233 kafka_publisher=kafka_broker.KafkaBrokerPublisher(parse_target)
234 for message in kafka_publisher.kafka_consumer:
235 #print message.value
236 #logging.info("%s",message.value)
237 process_notification_from_ceilometer(json.loads(message.value))
238 #print status
239 except Exception as e:
240 print "AUTO_SCALE Exception:",e
241'''
Matteo Scandolobf14f882016-06-02 10:01:34 -0700242def read_notification_from_ceilometer_over_kafka(host,port,topic):
243 print "Kafka target" , host, port, topic
244 try :
245 consumer=kafka.KafkaConsumer(bootstrap_servers=["%s:%s" % (host,port)])
raghunath dudyala01482a82016-08-01 12:29:16 +0530246 #consumer.assign([TopicPartition(topic,0)])
247 consumer.set_topic_partitions(topic)
Matteo Scandolobf14f882016-06-02 10:01:34 -0700248 consumer.seek_to_end()
249 for message in consumer:
raghunath dudyala01482a82016-08-01 12:29:16 +0530250 print message.value
Matteo Scandolobf14f882016-06-02 10:01:34 -0700251 #logging.debug("%s",message.value)
252 process_notification_from_ceilometer(json.loads(message.value))
253 #status = process_ceilometer_message(json.loads(message.value),message.value)
254 #print status
255 except Exception as e:
256 print "AUTO_SCALE Exception:",e
raghunath dudyala01482a82016-08-01 12:29:16 +0530257'''
Matteo Scandolobf14f882016-06-02 10:01:34 -0700258
259def read_notification_from_ceilometer(host,port):
260 udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP
261 udp.bind((host, port))
262
263 while True:
264 data, source = udp.recvfrom(64000)
265 try:
266 sample = msgpack.loads(data, encoding='utf-8')
267 process_notification_from_ceilometer(sample)
268 except Exception as e:
269 print e
270
271def process_notification_from_ceilometer(sample):
272 if sample['counter_name'] == 'instance':
273 if 'delete' in sample['resource_metadata']['event_type']:
274 xosTenantInfo = getXosTenantInfo(sample['project_id'])
275 xosResourceInfo = getXosInstanceInfo(sample['resource_id'])
276 print "SRIKANTH: Project %s Instance %s is getting deleted" % (xosTenantInfo['slice'] if xosTenantInfo['slice'] else sample['project_id'],xosResourceInfo)
277 if sample['project_id'] not in projects_map.keys():
278 return
279 if sample['resource_id'] not in projects_map[sample['project_id']]['resources'].keys():
280 return
281 projects_map[sample['project_id']]['resources'].pop(sample['resource_id'], None)
282 return
283 elif sample['counter_name'] != 'cpu_util':
284 return
285 if sample['project_id'] not in projects_map.keys():
286 projects_map[sample['project_id']] = {}
287 xosTenantInfo = getXosTenantInfo(sample['project_id'])
288 projects_map[sample['project_id']]['project_id'] = sample['project_id']
289 projects_map[sample['project_id']]['slice'] = xosTenantInfo['slice']
290 projects_map[sample['project_id']]['service'] = xosTenantInfo['service']
291 projects_map[sample['project_id']]['resources'] = {}
292 projects_map[sample['project_id']]['uthreadshold_count'] = 0
293 projects_map[sample['project_id']]['lthreadshold_count'] = 0
294 projects_map[sample['project_id']]['alarm'] = INITIAL_STATE
295 resource_map = projects_map[sample['project_id']]['resources']
296 if sample['resource_id'] not in resource_map.keys():
297 resource_map[sample['resource_id']] = {}
298 resource_map[sample['resource_id']]['xos_instance_info'] = getXosInstanceInfo(sample['resource_id'])
299 resource_map[sample['resource_id']]['queue'] = []
300 samples_queue = resource_map[sample['resource_id']]['queue']
301 sample = {'counter_name':sample['counter_name'],
302 'project_id':sample['project_id'],
303 'resource_id':sample['resource_id'],
304 'timestamp':sample['timestamp'],
305 'counter_unit':sample['counter_unit'],
306 'counter_volume':sample['counter_volume']}
307 deque = collections.deque(samples_queue, maxlen=10)
308 deque.append(sample)
309 resource_map[sample['resource_id']]['queue'] = list(deque)
310
311def setup_webserver():
312 try:
313 #config = ConfigParser.ConfigParser()
314 #config.read('pub_sub.conf')
315 #webserver_host = config.get('WEB_SERVER','webserver_host')
316 #webserver_port = int (config.get('WEB_SERVER','webserver_port'))
317 #client_host = config.get('CLIENT','client_host')
318 #client_port = int (config.get('CLIENT','client_port'))
319
320 #log_level = config.get('LOGGING','level')
321 #log_file = config.get('LOGGING','filename')
322
323 #level = LEVELS.get(log_level, logging.NOTSET)
324 #logging.basicConfig(filename=log_file,format='%(asctime)s %(levelname)s %(message)s',\
325 # datefmt=_DEFAULT_LOG_DATE_FORMAT,level=level)
326 webserver_host = '0.0.0.0'
327 webserver_port = 9991
328
329 except Exception as e:
330 print("* Error in config file:",e.__str__())
331 logging.error("* Error in confing file:%s",e.__str__())
332 else:
333 app.run(host=webserver_host,port=webserver_port,debug=True, use_reloader=False)
334
raghunath dudyala01482a82016-08-01 12:29:16 +0530335def parse_args(argv):
336 global xos_ip, kafka_ip
Matteo Scandolobf14f882016-06-02 10:01:34 -0700337
raghunath dudyala01482a82016-08-01 12:29:16 +0530338 try:
339 opts, args = getopt.getopt(argv,"k:x:",["kafka-ip=","xos-ip="])
340 except getopt.GetoptError:
341 print 'xos_auto_scaling_app.py --xos-ip=<IP> --kafka-ip=<>'
342 sys.exit(2)
343
344 for opt, arg in opts:
345 if opt in ("--xos-ip"):
346 xos_ip = arg
347 elif opt in ("--kafka-ip"):
348 kafka_ip = arg
349
350 if not xos_ip or not kafka_ip:
351 print 'xos_auto_scaling_app.py --xos-ip=<IP> --kafka-ip=<>'
352 sys.exit(2)
353
354
355def main(argv):
356 global xos_ip, kafka_ip
357
358 parse_args(argv)
Matteo Scandolobf14f882016-06-02 10:01:34 -0700359 monitoring_channel = acquire_xos_monitoring_channel()
360 if not monitoring_channel:
raghunath dudyala01482a82016-08-01 12:29:16 +0530361 print 'XOS monitoring_channel is not created... Create it before using this app'
Matteo Scandolobf14f882016-06-02 10:01:34 -0700362 return
363 loadAllXosTenantInfo()
364 loadAllXosInstanceInfo()
365 ceilometer_url = monitoring_channel['ceilometer_url']
366 if use_kafka:
raghunath dudyala01482a82016-08-01 12:29:16 +0530367 url = "kafka://" + kafka_ip + ":9092?topic=auto-scale"
368 parse_target=netutils.urlsplit(url)
369 thread.start_new(read_notification_from_ceilometer_over_kafka, (parse_target,))
370 #thread.start_new(read_notification_from_ceilometer_over_kafka, (xos_ip,"9092","auto-scale",))
371 subscribe_data = {"sub_info":"cpu_util","app_id":"xos_auto_scale","target":"kafka://" + kafka_ip + ":9092?topic=auto-scale"}
Matteo Scandolobf14f882016-06-02 10:01:34 -0700372 else:
373 thread.start_new(read_notification_from_ceilometer,(UDP_IP,UDP_PORT,))
raghunath dudyala01482a82016-08-01 12:29:16 +0530374 subscribe_data = {"sub_info":"cpu_util","app_id":"xos_auto_scale","target":"udp://" + xos_ip + ":12346"}
Matteo Scandolobf14f882016-06-02 10:01:34 -0700375 subscribe_url = ceilometer_url + 'v2/subscribe'
376 response = requests.post(subscribe_url, data=json.dumps(subscribe_data))
raghunath dudyala01482a82016-08-01 12:29:16 +0530377 print 'Ceilometer meter "cpu_util" Subscription status:%s' % response.text
Matteo Scandolobf14f882016-06-02 10:01:34 -0700378 #TODO: Fix the typo in 'sucess'
379 if (not 'sucess' in response.text) and (not 'already exists' in response.text):
raghunath dudyala01482a82016-08-01 12:29:16 +0530380 print 'Ceilometer meter "cpu_util" Subscription unsuccessful...Exiting'
Matteo Scandolobf14f882016-06-02 10:01:34 -0700381 return
raghunath dudyala01482a82016-08-01 12:29:16 +0530382 #TODO: Fix the need for different app_id for each subscription from the same application
383 if use_kafka:
384 subscribe_data = {"sub_info":"instance","app_id":"xos_auto_scale2","target":"kafka://" + kafka_ip + ":9092?topic=auto-scale"}
385 else:
386 subscribe_data = {"sub_info":"instance","app_id":"xos_auto_scale2","target":"udp://" + xos_ip + ":12346"}
Matteo Scandolobf14f882016-06-02 10:01:34 -0700387 subscribe_url = ceilometer_url + 'v2/subscribe'
388 response = requests.post(subscribe_url, data=json.dumps(subscribe_data))
raghunath dudyala01482a82016-08-01 12:29:16 +0530389 print 'Ceilometer meter "instance" Subscription status:%s' % response.text
Matteo Scandolobf14f882016-06-02 10:01:34 -0700390 #TODO: Fix the typo in 'sucess'
391 if (not 'sucess' in response.text) and (not 'already exists' in response.text):
raghunath dudyala01482a82016-08-01 12:29:16 +0530392 print 'Ceilometer meter "instance"Subscription unsuccessful...Exiting'
Matteo Scandolobf14f882016-06-02 10:01:34 -0700393 return
394 periodic_cpu_threshold_evaluator()
395 periodic_print()
396 setup_webserver()
397
398if __name__ == "__main__":
raghunath dudyala01482a82016-08-01 12:29:16 +0530399 main(sys.argv[1:])