Merge pull request #3 from zorroblue/crawler

Added crawler

Author: zorroblue

CONTRIBUTING.md

@@ -47,3 +47,6 @@
 
  	python -m grpc_tools.protoc -I./protos --python_out=. --grpc_python_out=. protos/search.proto
 
+ #### Defaults:
+
+ 	The master server runs in port 50051 while replica runs in port 50052. The database name of master is 'masterdb' while replica's is 'replicadb'

crawler.py

@@ -0,0 +1,165 @@
+from concurrent import futures
+import time
+import math
+
+from argparse import ArgumentParser
+import argparse
+
+import grpc
+import search_pb2
+import search_pb2_grpc
+
+import json
+import logging
+
+from utils import querydb, init_logger, parse_level
+from data.generatedata import generate_indices
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
+
+def build_parser():
+	parser = ArgumentParser()
+	parser.add_argument('--master',
+			dest='master', help='Master IP address',
+			default='localhost:50051',
+			required=False)
+	parser.add_argument('--replica',
+			dest='replica',
+			default='localhost:50052',
+			help='Replica IP address',
+			required=False)
+	choices = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
+	parser.add_argument('--logging',
+			dest='logging_level', help='Logging level',
+			choices=choices,
+			default='DEBUG',
+			required=False)
+	return parser
+
+
+def write_to_master_replica(master, replica, logging_level, lower=51, higher=70):
+	# initialize logger
+	logger = init_logger('crawler', logging_level)
+	# generate data
+	indices = generate_indices('pending', lower, higher)
+	indices = json.dumps(indices)
+	# initiate 2 phase commit protocol
+	
+	# PHASE 1
+	logger.debug("Initiate 2 phase commit protocol")
+	print "Phase 1: Prepare"
+	logger.debug("Starting Phase 1: Prepare")
+
+	master_vote = None
+	replica_vote = None
+	
+	master_channel = grpc.insecure_channel(master)
+	master_stub = search_pb2_grpc.DatabaseWriteStub(master_channel)
+	
+	request = search_pb2.CommitRequest(data=indices)
+
+	replica_channel = grpc.insecure_channel(replica)
+	replica_stub = search_pb2_grpc.DatabaseWriteStub(replica_channel)
+
+	try:
+		logger.info("Send COMMIT_REQUEST to master")
+		master_vote = master_stub.QueryToCommit(request)
+		print "Master Status: ", master_vote.status
+		if master_vote.status == 1:
+			logger.info("Received AGREED from master")
+		else:
+			logger.info("Received ABORT from master")
+	except Exception as e:
+		print e.code()
+		logger.error("Master not reachable due to "+ str(e.code()))
+
+	try:
+		logger.info("Send COMMIT_REQUEST to backup")
+		replica_vote = replica_stub.QueryToCommit(request)
+		print "Backup Status: ", replica_vote.status
+		if replica_vote.status == 1:
+			logger.info("Received AGREED from backup")
+		else:
+			logger.info("Received ABORT from backup")
+	except Exception as e:
+		print e.code()
+		logger.error("Backup not reachable due to "+ str(e.code()))
+	
+	# PHASE 2
+	print "Phase 2: Prepare"
+	logger.debug("Starting Phase 2: Commit")
+
+	if master_vote == None or replica_vote == None or replica_vote.status == 0 or master_vote.status == 0:
+		try:
+			request = search_pb2.CommitStatusUpdate(code=search_pb2.ROLL_BACK)
+			try:
+				logger.info("Sending ROLL_BACK to master")
+				master_ack = master_stub.CommitPhase(request)
+			except Exception as e:
+				print str(e)
+				logger.info("Master not able to receive ROLL_BACK due to "+ str(e.code()))
+			
+			try:
+				logger.info("Sending ROLL_BACK to replica")
+				replica_ack = replica_stub.CommitPhase(request)
+			except Exception as e:
+				print str(e)
+				logger.info("Replica not able to receive ROLL_BACK due to "+ str(e.code()))
+
+		except Exception as e:
+			print e.code()
+		logger.info("Rolled back transaction")
+		return False
+
+	# Commit Phase
+	request = search_pb2.CommitStatusUpdate(code=search_pb2.COMMIT)
+	master_ack_received = False
+	replica_ack_received = False
+	retries = 0
+	
+	while (not master_ack_received or not replica_ack_received) and retries < 3: 
+		if retries > 0:
+			logger.info("Retrying")
+		if not master_ack_received:
+			try:
+				logger.info("Sending COMMIT to master")
+				master_ack = master_stub.CommitPhase(request)
+				master_ack_received = True
+			except Exception as e:
+				logger.info("Master failed to receive due to "+ str(e.code()))
+		if not replica_ack_received:
+			try:
+				logger.info("Sending COMMIT to backup")
+				replica_ack = replica_stub.CommitPhase(request)
+				replica_ack_received = True
+			except Exception as e:
+				logger.info("Backup failed to receive due to "+ str(e.code()))
+		retries+=1
+		print master_ack_received, replica_ack_received
+		
+
+	if retries < 3:
+		logger.info("COMMIT")
+		# TODO rollback 
+		# not doing it now to progress further in the work
+		# PS if one of them fails, they will resync with the crawler. The crawler is aware of this as COMMIT isn't written till then.
+
+	print "Phase 2:"
+	print "Master status", master_ack.status
+	print "Replica status", replica_ack.status
+	return master_ack.status == 1 and replica_ack.status == 1
+
+
+def main():
+	parser = build_parser()
+	options = parser.parse_args()
+	master = options.master
+	replica = options.replica
+	logging_level = parse_level(options.logging_level)
+	status = write_to_master_replica(master=master, replica=replica, logging_level=logging_level)
+	if status:
+		print "Write successful"
+
+if __name__ == '__main__':
+	main()

data/__init__.py

@@ -2,48 +2,69 @@
 from random_words import RandomWords
 import random
 from datetime import datetime
+import json
+import pkg_resources
 
 random.seed(datetime.now())
-rw = RandomWords()
-
-f = open("synonyms.txt")
 
-indices = {}
-linecount = 0
 
 def generate_random_urls(word):
 	'''Generate random urls for a word
 	For any word, pick a random number of urls to return 
 	and choose random urls starting with the second letter of the word
 	'''
+	rw = RandomWords()
 	r = random.randint(2,6)
 	words = rw.random_words(count=r, letter=word[0])
 	urls = []
 	for word in words:
 		url = "https://www."+word+".com"
 		urls.append(url)
-	print urls
+	#print urls
 	return urls
 
-for line in f:
-	words = line.strip().split(",")
-	for i in range(len(words)):
-		words[i] = words[i].strip()
-	linecount += 1
-	# pick only first 50 sets
-	if linecount > 50:
- 		break
-	for word in words:
-		word = word.strip()
-		indices[word] = {}
-		indices[word]['sim_words'] = words
-		indices[word]['urls'] = generate_random_urls(word)
+def generate_indices(status='committed', lower_range=1, upper_range=50):
+	resource_package = __name__  # Could be any module/package name
+	resource_path = 'synonyms.txt'  # Do not use os.path.join(), see below
+	f = pkg_resources.resource_stream(resource_package, resource_path)
+	
+	indices = {}
+	linecount = 0
+	for line in f:
+		words = line.strip().split(",")
+		for i in range(len(words)):
+			words[i] = words[i].strip()
+		linecount += 1
+		# pick only first sets in lines [lower_range, upper range]
+		if linecount < lower_range:
+			continue
+		if linecount > upper_range:
+	 		break
+		for word in words:
+			word = word.strip()
+			if len(word) == 0:
+				continue
+			indices[word] = {}
+			indices[word]['status'] = status
+			indices[word]['sim_words'] = words
+			indices[word]['urls'] = generate_random_urls(word)
+		json_list = []
+
+		for word in indices:
+			indices[word]['name'] = word
+			json_list.append(indices[word])
+	return json_list
+
+def main():
+	indices = generate_indices()
+	json_list = []
 
-json_list = []
+	for word in indices:
+		indices[word]['name'] = word
+		json_list.append(indices[word])
 
-for word in indices:
-	indices[word]['name'] = word
-	json_list.append(indices[word])
+	with open('indices.json', 'w') as fp:
+		json.dump(json_list, fp, sort_keys=True, indent=4)
 
-with open('indices.json', 'w') as fp:
-	json.dump(json_list, fp, sort_keys=True, indent=4)
+if __name__ == '__main__':
+	main()