In this assignment, we will use the Sanketplus/PyDFS library in Github (see link below) to make a distributed file system application. Be sure to review link below to become familiar with this code.
You will first need to create a minion node (see code below). Use this code to create two different minion nodes. Be sure to review the lecture video in Teams as well.
import rpyc
import os
import sys
import logging
from rpyc.utils.server import ThreadedServer
DATA_DIR = "/tmp/minion/"
PORT = 8888
logging.basicConfig(level=logging.DEBUG)
class Minion(rpyc.Service):
def exposed_put(self, block_id, data, minions):
logging.debug("put block: " + block_id)
out_path = os.path.join(DATA_DIR, block_id)
with open(out_path, 'w') as f:
f.write(data)
if len(minions) > 0:
self.forward(block_id, data, minions)
def exposed_get(self, block_id):
logging.debug("get block: " + block_id)
block_addr = os.path.join(DATA_DIR, block_id)
if not os.path.isfile(block_addr):
logging.debug("block not found")
return None
with open(block_addr) as f:
return f.read()
def forward(self, block_id, data, minions):
logging.debug("forwarding block: " + block_id + str(minions))
next_minion = minions[0]
minions = minions[1:]
host, port = next_minion
rpyc.connect(host, port=port).root.put(block_id, data, minions)
if __name__ == "__main__":
PORT = int(sys.argv[1])
DATA_DIR = sys.argv[2]
if not os.path.isdir(DATA_DIR):
os.mkdir(DATA_DIR)
logging.debug("starting minion")
rpyc_logger = logging.getLogger('rpyc')
rpyc_logger.setLevel(logging.WARN)
t = ThreadedServer(Minion(), port=PORT, logger=rpyc_logger, protocol_config={
'allow_public_attrs': True,
})
t.start()
Next, you will need to start the master node.
import rpyc
import uuid
import math
import random
import configparser
import signal
import pickle
import sys
import os
from rpyc.utils.server import ThreadedServer
BLOCK_SIZE = 100
REPLICATION_FACTOR = 2
MINIONS = {"1": ("127.0.0.1", 8000),
"2": ("127.0.0.1", 9000),}
class MasterService(rpyc.Service):
"""
file_block = {'file.txt': ["block1", "block2"]}
block_minion = {"block1": [1,3]}
minions = {"1": (127.0.0.1,8000), "3": (127.0.0.1,9000)}
"""
file_block = {}
block_minion = {}
minions = MINIONS
block_size = BLOCK_SIZE
replication_factor = REPLICATION_FACTOR
def exposed_read(self, file):
mapping = []
# iterate over all of file's blocks
for blk in self.file_block[file]:
minion_addr = []
# get all minions that contain that block
for m_id in self.block_minion[blk]:
minion_addr.append(self.minions[m_id])
mapping.append({"block_id": blk, "block_addr": minion_addr})
return mapping
def exposed_write(self, file, size):
self.file_block[file] = []
num_blocks = int(math.ceil(float(size) / self.block_size))
return self.alloc_blocks(file, num_blocks)
def alloc_blocks(self, file, num_blocks):
return_blocks = []
for i in range(0, num_blocks):
block_id = str(uuid.uuid1()) # generate a block
minion_ids = random.sample( # allocate REPLICATION_FACTOR number of minions
list(self.minions.keys()), self.replication_factor)
minion_addr = [self.minions[m] for m in minion_ids]
self.block_minion[block_id] = minion_ids
self.file_block[file].append(block_id)
return_blocks.append(
{"block_id": block_id, "block_addr": minion_addr})
return return_blocks
if __name__ == "__main__":
t = ThreadedServer(MasterService(), port=2131, protocol_config={
'allow_public_attrs': True,
})
t.start()
Finally, the follow code will allow you to query the distributed file system (see below).
import rpyc
import sys
import os
import logging
logging.basicConfig(level=logging.DEBUG)
def get(master, file):
file_table = master.read(file)
if not file_table:
logging.info("file not found")
return
for block in file_table:
for host, port in block['block_addr']:
try:
con = rpyc.connect(host, port=port).root
data = con.get(block['block_id'])
if data:
sys.stdout.write(data)
break
except Exception as e:
continue
else:
logging.error("No blocks found. Possibly a corrupt file")
def put(master, source, dest):
size = os.path.getsize(source)
blocks = master.write(dest, size)
with open(source) as f:
for block in blocks:
data = f.read(master.block_size)
block_id = block['block_id']
minions = block['block_addr']
minion = minions[0]
minions = minions[1:]
host, port = minion
con = rpyc.connect(host, port=port)
con.root.put(block_id, data, minions)
def main(args):
con = rpyc.connect("localhost", port=2131)
master = con.root
if args[0] == "get":
get(master, args[1])
elif args[0] == "put":
put(master, args[1], args[2])
else:
logging.error("try 'put srcFile destFile OR get file'")
if __name__ == "__main__":
main(sys.argv[1:])
One of the problems with this code from Github is that it won’t save state between stopping and starting the Master file. For this assignment, you will need to implement that functionality. Use the two files in the zip file in Canvas to track the internal state (“PyDFS_Master_File_Block_File.dat” and “PyDFS_Master_Block_Minion_File.dat”). The first file will hold the dictionary contents of the file_block dictionary in the Master program. The second file will hold the dictionary contents of the block_minion dictionary in the Master program. Both are needed to implement the desired functionality.
(Hint: the only file you need to modify to implement this functionality is the Master program. Also, pay close attention to what the alloc_blocks function is doing in the Master file. This will be key. You may also want to use the “ast” library in Python to assist with reading a file into a dictionary.)
****Be sure to include screenshots of your working code in addition to submitting the Python code itself.
Also, you must implement this functionality using the two files I mentioned previously (e.g., “PyDFS_Master_File_Block_File.dat” and “PyDFS_Master_Block_Minion_File.dat”) in order to receive credit for this assignment.
Submit your Python source files and any other relevant material to Canvas.
CS 340 Milestone One Guidelines and Rubric Overview: For this assignment, you will implement the fundamental operations of create, read, update,
Retail Transaction Programming Project Project Requirements: Develop a program to emulate a purchase transaction at a retail store. This
7COM1028 Secure Systems Programming Referral Coursework: Secure
Create a GUI program that:Accepts the following from a user:Item NameItem QuantityItem PriceAllows the user to create a file to store the sales receip
CS 340 Final Project Guidelines and Rubric Overview The final project will encompass developing a web service using a software stack and impleme
