text: #   Copyright 2010-2012 Opera Software ASA 
  #
  #   Licensed under the Apache License, Version 2.0 (the "License");
  #   you may not use this file except in compliance with the License.
  #   You may obtain a copy of the License at
  #
  #       http://www.apache.org/licenses/LICENSE-2.0
  #
  #   Unless required by applicable law or agreed to in writing, software
  #   distributed under the License is distributed on an "AS IS" BASIS,
  #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #   See the License for the specific language governing permissions and
  #   limitations under the License.
  
  
  import libinit
  import probedb.standalone
  import probedb.cluster.models as Cluster
  import probedb.probedata2.models as Probe
  import os,os.path,time,subprocess,datetime
  from optparse import OptionParser
  from probedb.cluster.models import ClusterAction
  from django.db import connection
  
  """Main handler on each cluster, checks every 10 minutes for new tasks"""
  
  computername =  os.environ.get('COMPUTERNAME',"any").lower()
  if computername == "any":
  	computername =  os.environ.get('HOSTNAME',"any").lower()
  if computername == "any":
  	raise Exception("Computername was empty")
  
  computername = computername.partition('.')[0]
  
  options_config = OptionParser()
  
  options_config.add_option("--testbase2", action="store_true", dest="use_testbase2")
  options_config.add_option("--verbose", action="store_true", dest="verbose")
  options_config.add_option("--managed", action="store_true", dest="managed")
  options_config.add_option("--performance", action="store_true", dest="register_performance")
  
  (options, args) = options_config.parse_args()
  
  
  while True:
  	terminate = False
  	master_configuration,created = Cluster.ClusterNode.objects.get_or_create(hostname = "tlsprober-cluster", defaults={
  																				"probe_parameters":"not used",
  																				#"result_parameters":"not used",
  																				"active_node":True,
  																				})
  	configuration,created = Cluster.ClusterNode.objects.get_or_create(hostname = computername, defaults={
  																				"probe_parameters":"--processes 40 --iterations 40",
  																				#"result_parameters":"--processes 10 --iterations 100",
  																				"active_node":True,
  																				})
  	configuration.save()
  
  	for run in Cluster.ClusterRun.objects.filter(enabled=True).order_by("-priority", "entered_date").iterator():
  		"""Find the active job with the highest priority in the queue"""  
  		if run.perform_run.probequeue_set.filter(state=Probe.ProbeQueue.PROBEQ_IDLE).count() == 0:
  			run.enabled = False
  			run.save()
  			continue
  		
  		if run.perform_run.branch:
  			terminate = True
  		
  		processes = []
  		process_index = 0
  		topped_out = False
  		
  		check_queue_frequency = 1
  		checked_count = 0
  		checked_count_git = 0
  		last_count =Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id,state=Probe.ProbeQueue.PROBEQ_IDLE).count()
  		if last_count > 1000000:
  			check_queue_frequency = 30			
  		elif last_count > 50000:
  			check_queue_frequency = 10			
  		
  		while (Cluster.ClusterRun.objects.filter(enabled=True, priority__gt=run.priority).count() == 0 and 
  			Cluster.ClusterRun.objects.get(id = run.id).enabled):
  			master_configuration = Cluster.ClusterNode.objects.get(hostname = "tlsprober-cluster")
  			configuration= Cluster.ClusterNode.objects.get(hostname = computername)
  			if not configuration.active_node or not master_configuration.active_node:
  				break;
  			configuration.save()
  
  			if not processes:
  				subprocess.call(["git", "pull",])
  				subprocess.call(["git", "submodule", "update", "--recursive"])
  				subprocess.call(["python", "build_certhandler.py", "build","--build-lib","."], cwd=os.path.join("probedb","certs"))
  
  			checked_count += 1
  			if checked_count >= check_queue_frequency:
  				qlen =Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id, state=Probe.ProbeQueue.PROBEQ_IDLE).count()
  				if qlen <= 0 :
  					break;
  				if qlen < 50000:
  					check_queue_frequency = 0
  				elif qlen < 1000000:
  					check_queue_frequency = 10
  					
  				last_count = qlen			
  
  				checked_count = 0
  				
  			checked_count_git += 1;
  			if checked_count_git>= 10:
  				subprocess.call(["git", "pull",])
  				subprocess.call(["git", "submodule", "update", "--recursive"])
  				subprocess.call(["python", "build_certhandler.py", "build","--build-lib","."], cwd=os.path.join("probedb","certs"))
  				checked_count_git = 0;
  
  			
  			run_config = OptionParser()
  			
  			run_config.add_option("--processes", action="store", type="int", dest="process_count", default=1)
  			run_config.add_option("--iterations", action="store", type="int", dest="iteration_count", default=40)
  
  			(run_options, args) = run_config.parse_args(configuration.probe_parameters.split())
  			if int(run_options.process_count) == 0:
  				break
  
  			started_proc = 0;
  			
  			Q = Cluster.ClusterAction.objects.filter(cluster_run=run)
  			current_time = datetime.datetime.now()
  			delta = datetime.timedelta(minutes=10)
  
  			# Calculate process and iteration limits based on the actual 
  			# number of items left for the job
  			# Early in the process the cluster will work slowly up to full 
  			# number of processes, to avoid overloading the database
  			# towards the end, the number of processes will be reduced,
  			# in parallel to reducing the number of iterations per process
  			# to reduce the risk that one process will continue for a long 
  			# time while others have finished  
  			Q3 = Q.filter(completed_time__range=(current_time-delta,current_time-datetime.timedelta(minutes=5)))
  			all_count_5min = Q3.count()
  			if all_count_5min > 4:
  			
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				node_count_10min = Q1.filter(cluster_node = configuration).count()
  				all_count_10min = Q1.count()
  				
  				if node_count_10min > 4:
  					factor = all_count_10min/node_count_10min
  					iterations = int((last_count * factor)/ int(run_options.process_count))
  
  					if run_options.iteration_count > iterations:
  						run_options.iteration_count = max(15, iterations)
  						
  			proc_limit = int(run_options.process_count)
  			if last_count < 10000 and proc_limit > 20 and run_options.iteration_count >50 :
  				run_options.iteration_count = 15
  				
  			if last_count < 100000 and proc_limit > 50:
  				proc_limit /= 3
  				proc_limit = max(40,proc_limit)
  				if last_count < run_options.iteration_count * proc_limit:
  					proc_limit = max(min(40,proc_limit), min(40,last_count/ run_options.iteration_count, proc_limit))
  
  			if last_count < proc_limit*run_options.iteration_count:
  				proc_limit = max(10, last_count/run_options.iteration_count)
  
  			if proc_limit > 80 and len(processes) >0 and  len(processes)<80 and all_count_5min < 50:
  				time.sleep(120);
  				continue;
  
  			start_proc = max(10,min(
  								30 if proc_limit *0.75 > len(processes) or (proc_limit > 200 and len(processes) < 100)  else 50 , 
  								proc_limit/10))
  			if proc_limit >80 and len(processes)<30:
  				start_proc = 5
  			while len(processes) < proc_limit and started_proc<start_proc:
  				process_index += 1
  				new_process = subprocess.Popen(
  											(["nice"] if os.name == "posix" else []) +
  											["python", "-O", 
  												"probe_lists.py",
  												"-n",  str(process_index),
  												"--run-id", str(run.perform_run.id),
  												"--source", '"'+run.perform_run.source_name+'"',
  												"--description", '"'+run.perform_run.description+'"',
  												"--max", str(run_options.iteration_count),
  												]+
  												(["--performance"] if options.verbose  or options.register_performance else [])+
  												(["--testbase2"] if options.use_testbase2 else [])+
  												(["--large_run"] if last_count > 400000 else [])+
  												(["--small_run"] if last_count < 50000 else [])
  												, shell=False)
  				
  				started_proc += 1
  				if new_process:
  					processes.append(new_process)
  					if options.verbose:
  						print "started ", process_index, " count ", len(processes), "/", proc_limit
  				time.sleep(0.5)
  		
  			try:
  				connection.close()
  			except:
  				pass
  			
  			if not topped_out and len(processes) >= int(run_options.process_count):
  				topped_out = True
  			
  			# Don't start too processes to quickly
  			if (start_proc < 10 and not topped_out) or (proc_limit > 200 and len(processes) < proc_limit*0.90):
  				time.sleep(300)
  			elif last_count < 100000 and proc_limit > 20 and not topped_out:
  				time.sleep(240)
  			elif len(processes) >300 and len(processes) < proc_limit*0.95:
  				time.sleep(300)
  			elif len(processes) >200 and len(processes) < proc_limit*0.95:
  				time.sleep(60)
  			else:
  				time.sleep(30 if len(processes) > proc_limit*0.95 else 15)
  			
  			if options.verbose:
  				Q = ClusterAction.objects.filter(cluster_run=run)
  				current_time = datetime.datetime.now()
  				delta = datetime.timedelta(hours=1)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_hour = Q1.count() 
  				node_count_hour = Q1.filter(cluster_node = configuration).count() 
  
  				delta = datetime.timedelta(minutes=10)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_10min = Q1.count()*6 #recalc to per hour 
  				node_count_10min = Q1.filter(cluster_node = configuration).count()*6  #recalc to per hour
  
  				delta = datetime.timedelta(minutes=1)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_1min = Q1.count()*60 #recalc to per hour 
  				node_count_1min = Q1.filter(cluster_node = configuration).count()*60  #recalc to per hour
  				
  				#clean up queue
  				#Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id, state=Probe.ProbeQueue.PROBEQ_FINISHED).delete()
  
  				print "Node: ", node_count_1min, " ",  node_count_10min, " ", node_count_hour
  				print "All: ", all_count_1min, " ",  all_count_10min, " ", all_count_hour, " (", Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id,state=Probe.ProbeQueue.PROBEQ_IDLE).count(), " left)"
  			
  			next_process_list = []
  			
  			for proc in processes:
  				if proc.poll() == None:
  					next_process_list.append(proc)
  			
  			processes = next_process_list
  				
  			# Loop back and try the next one
  		
  		while processes:
  			try:
  				connection.close()
  			except:
  				pass
  			time.sleep(30)
  			if options.verbose:
  				Q = ClusterAction.objects.filter(cluster_run=run)
  				current_time = datetime.datetime.now()
  				delta = datetime.timedelta(hours=1)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_hour = Q1.count() 
  				node_count_hour = Q1.filter(cluster_node = configuration).count() 
  
  				delta = datetime.timedelta(minutes=10)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_10min = Q1.count()*6 #recalc to per hour 
  				node_count_10min = Q1.filter(cluster_node = configuration).count()*6  #recalc to per hour
  
  				delta = datetime.timedelta(minutes=1)
  				Q1 = Q.filter(completed_time__range=(current_time-delta,current_time))
  				all_count_1min = Q1.count()*60 #recalc to per hour 
  				node_count_1min = Q1.filter(cluster_node = configuration).count()*60  #recalc to per hour
  				
  				#clean up queue
  				#Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id, state=Probe.ProbeQueue.PROBEQ_FINISHED).delete()
  
  				print "Node: ", node_count_1min, " ",  node_count_10min, " ", node_count_hour
  				print "All: ", all_count_1min, " ",  all_count_10min, " ", all_count_hour, " (", Probe.ProbeQueue.objects.filter(part_of_run=run.perform_run_id, state=Probe.ProbeQueue.PROBEQ_IDLE).count(), " left)"
  
  			next_process_list = []
  			
  			for proc in processes:
  				if proc.poll() == None:
  					next_process_list.append(proc)
  			
  			processes = next_process_list
  			
  			if options.verbose:
  				print "closing down: count ", len(processes), "/",  run_options.process_count
  				
  			# Loop back and see if all has ended now
  		
  		
  		if options.verbose:
  			print "closed down:"
  			
  		break; # need to trigger a new database retrieval
  	
  	if terminate:
  		break;
  	
  	master_configuration = Cluster.ClusterNode.objects.get(hostname = "tlsprober-cluster")
  	configuration= Cluster.ClusterNode.objects.get(hostname = computername)
  	configuration.save()
  	run_config = OptionParser()
  	
  	run_config.add_option("--processes", action="store", dest="process_count", default=1)
  	run_config.add_option("--iterations", action="store", dest="iteration_count", default=40)
  
  	(run_options, args) = run_config.parse_args(configuration.probe_parameters.split())
  	if (not master_configuration.active_node or not configuration.active_node or 
  		int(run_options.process_count) == 0 or Cluster.ClusterRun.objects.filter(enabled=True).count() == 0):
  		if options.managed and Cluster.ClusterRun.objects.filter(enabled=True).count() == 0:
  			break;
  		connection.close()
  		time.sleep(60) # if there are no runs, sleep for 10 minutes
  