Skip to content

Sourcery refactored master branch #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sourcery-ai wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Sourcery refactored master branch #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions experiments/request_sender.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,7 @@ def get_data():
with open(f'{upper_folder}/resources/test.jpg', 'rb') as f:
raw_data = f.read()
base64_bytes = b64encode(raw_data)
base64_string = base64_bytes.decode('utf-8')
return base64_string
return base64_bytes.decode('utf-8')

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function get_data refactored with the following changes:


def send_data(args, reader):
pool = ThreadPoolExecutor(5000)
Expand All @@ -61,8 +60,7 @@ def get_kr_data():
with open(test_file, 'rb') as f:
raw_data = f.read()
base64_bytes = b64encode(raw_data)
base64_string = base64_bytes.decode('utf-8')
return base64_string
return base64_bytes.decode('utf-8')
Comment on lines -64 to +63

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function get_kr_data refactored with the following changes:


def send_data_kr(args, reader):
pool = ThreadPoolExecutor(5000)
Expand All @@ -72,7 +70,7 @@ def send_data_kr(args, reader):
if reader.line_num > args.timeout:
break

num = int(int(row['tweets']) / 2)
num = int(row['tweets']) // 2
Comment on lines -75 to +73

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function send_data_kr refactored with the following changes:

lam = (60 * 1000.0) / num
samples = np.random.poisson(lam, num)
print(f'line: {reader.line_num}; sample_number: {num}')
Expand All @@ -93,7 +91,7 @@ def send_data_nmt(args, reader):
if reader.line_num > args.timeout:
break

num = int(int(row['tweets']) / 2)
num = int(row['tweets']) // 2
Comment on lines -96 to +94

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function send_data_nmt refactored with the following changes:

lam = (60 * 1000.0) / num
samples = np.random.poisson(lam, num)
print(f'line: {reader.line_num}; sample_number: {num}')
Expand Down Expand Up @@ -128,8 +126,8 @@ def send_data_mx(args, reader):
def send_fixed_data(args):
pool = ThreadPoolExecutor(3000)
data = get_data()
num = 3000
for _ in range(20):
num = 3000
Comment on lines +129 to -132

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function send_fixed_data refactored with the following changes:

lam = (60 * 1000.0) / num
samples = np.random.poisson(lam, num)
for s in samples:
Expand Down
6 changes: 1 addition & 5 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,8 @@ def get_args():
return parser.parse_args()

def copy_keys(args, keys):
dst_dict = {}
src_dict = vars(args)
for key in keys:
if src_dict[key] is not None:
dst_dict[key] = src_dict[key]
return dst_dict
return {key: src_dict[key] for key in keys if src_dict[key] is not None}
Comment on lines -27 to +28

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function copy_keys refactored with the following changes:


"""
port -> just port as we know
Expand Down
150 changes: 74 additions & 76 deletions modules/aws_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,35 +23,35 @@

@app.task
def check_spot_states():
all_info = aws_accessor.get_all_cluster()
if all_info:
client = get_client()
spot_list = []
[ spot_list.append(c) for c in all_info ]

cancel_reqs = []
num = 0
for info in spot_list:
name = info['name']
for req, i in info['info'].items():
res = client.describe_spot_fleet_requests(SpotFleetRequestIds=[req])
if res['SpotFleetRequestConfigs'][0]['SpotFleetRequestState'] != 'active':
logging.info(f'Inactive spot request detected: {req}')
num += len(i['instance_id_list'])
cancel_reqs.append(req)
start_on_demand_instances(name)
if not (all_info := aws_accessor.get_all_cluster()):
return
client = get_client()
spot_list = []
[ spot_list.append(c) for c in all_info ]

cancel_reqs = []
num = 0
for info in spot_list:
name = info['name']
for req, i in info['info'].items():
res = client.describe_spot_fleet_requests(SpotFleetRequestIds=[req])
if res['SpotFleetRequestConfigs'][0]['SpotFleetRequestState'] != 'active':
logging.info(f'Inactive spot request detected: {req}')
num += len(i['instance_id_list'])
cancel_reqs.append(req)
start_on_demand_instances(name)

time.sleep(10)
if len(cancel_reqs) > 0:
logging.info('Cancel inactive requests')
cancel_spot_instances(name, cancel_reqs)
time.sleep(10)
if cancel_reqs:
logging.info('Cancel inactive requests')
cancel_spot_instances(name, cancel_reqs)

# interrupt recovery test
time.sleep(100)
logging.info(f'Monitor launching {num} spot instances')
name = 'inception'
launch_spot_instances(name, {'imageId':AMIS[DEFAULT_REGION]['CPU'], 'instanceType':'c5.large', 'targetCapacity':num, 'key_value':[('exp_round', 0)] })
stop_on_demand_instances(name)
# interrupt recovery test
time.sleep(100)
logging.info(f'Monitor launching {num} spot instances')
name = 'inception'
launch_spot_instances(name, {'imageId':AMIS[DEFAULT_REGION]['CPU'], 'instanceType':'c5.large', 'targetCapacity':num, 'key_value':[('exp_round', 0)] })
stop_on_demand_instances(name)
Comment on lines -26 to +54

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function check_spot_states refactored with the following changes:


@app.task
def launch_on_demand_instances(name, params):
Expand Down Expand Up @@ -91,27 +91,25 @@ def launch_on_demand_instances(name, params):

if 'key_value' in params:
_add_tags(ec2.meta.client, ids, params['key_value'])


ins_list = utils.get_ins_from_ids(params['region'], ids)
# check the state of instances by trying ssh
while True:
logging.info('Checking SSH connection')
if all([_check_ssh(i.ip) for i in ins_list]):
if all(_check_ssh(i.ip) for i in ins_list):
logging.info('Checked SSH connection')
break
time.sleep(10)
ins_list = utils.get_ins_from_ids(params['region'], ids)

mdl_source.setup_config(ins_list, params['region'], params['instanceType'])

ins_dict = {}
[ ins_dict.update( {i: utils.get_ins_from_id(ec2, params['region'], i)} ) for i in ids]

pre_demand_aws_accessor.del_requests(name, ids)
save_ins = {}
for id, ins in ins_dict.items():
save_ins[id] = ins.__dict__
save_ins = {id: ins.__dict__ for id, ins in ins_dict.items()}
Comment on lines -94 to +112

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function launch_on_demand_instances refactored with the following changes:

demand_aws_accessor.save_cluster(name, save_ins)

def stop_on_demand_instances(name, typ='t2.medium' , region=DEFAULT_REGION):
Expand All @@ -120,36 +118,35 @@ def stop_on_demand_instances(name, typ='t2.medium' , region=DEFAULT_REGION):
ins = []
[ins.append(id) for id, instance in info.items() if instance['typ'] == typ and instance['region'] == region]

if len(ins) > 0:
if ins:
logging.info(f'Stop {len(ins)} {typ} on demand backup instances')
backup_ins_accessor.del_all_instance()
res = client.stop_instances(InstanceIds=ins)
while True:
res = client.describe_instances(InstanceIds=ins)
ins_res = res['Reservations'][0]['Instances']
if all([ r['State']['Name'] == 'stopped' for r in ins_res ]):
logging.info(f'Backup instances stopped')
if all(r['State']['Name'] == 'stopped' for r in ins_res):
logging.info('Backup instances stopped')
break
logging.info(f'Checking backup instances state')
logging.info('Checking backup instances state')
Comment on lines -123 to +131

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function stop_on_demand_instances refactored with the following changes:

time.sleep(10)


def start_on_demand_instances(name, typ='t2.medium' , region=DEFAULT_REGION):
client = get_client()
record = demand_aws_accessor.get_cluster(name)
if record:
if record := demand_aws_accessor.get_cluster(name):
info = record['info']
ins = []
[ins.append(id) for id, instance in info.items() if instance['typ'] == typ and instance['region'] == region]

if len(ins) > 0:
if ins:
logging.info(f'Start {len(ins)} {typ} on demand instances')
res = client.start_instances(InstanceIds=ins)
instances = utils.get_ins_from_ids(region, ins)

while True:
logging.info('Checking SSH connection')
if all([_check_ssh(i.ip) for i in instances]):
if all(_check_ssh(i.ip) for i in instances):
Comment on lines -139 to +149

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function start_on_demand_instances refactored with the following changes:

logging.info('Checked SSH connection')
break
time.sleep(10)
Expand Down Expand Up @@ -182,15 +179,14 @@ def kill_on_demand_instances(name, region, typ, num):
def kill_all_on_demand_ins(name, region):
ec2 = boto3.resource('ec2', region_name=region, **CREDENTIALS)
records = demand_aws_accessor.get_cluster(name)
pre_rec = pre_demand_aws_accessor.get_cluster(name)
if pre_rec:
if pre_rec := pre_demand_aws_accessor.get_cluster(name):
pre_ids = pre_rec['info'].keys()
if len(list(pre_ids)) > 0:
if list(pre_ids):
ec2.instances.filter(InstanceIds=list(pre_ids)).terminate()
pre_demand_aws_accessor.del_requests(name, pre_ids)
if records:
ids = records['info'].keys()
if len(list(ids)) > 0:
if list(ids):
Comment on lines -185 to +189

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function kill_all_on_demand_ins refactored with the following changes:

ec2.instances.filter(InstanceIds=list(ids)).terminate()
demand_aws_accessor.del_requests(name, ids)

Expand Down Expand Up @@ -230,7 +226,7 @@ def launch_spot_instances(name, params):
# check the state of instances by trying ssh
while True:
logging.info('Checking SSH connection')
if all([_check_ssh(i.ip) for i in ins]):
if all(_check_ssh(i.ip) for i in ins):
Comment on lines -233 to +229

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function launch_spot_instances refactored with the following changes:

logging.info('Checked SSH connection')
break
time.sleep(10)
Expand Down Expand Up @@ -267,7 +263,7 @@ def kill_spot_instances_by_num(name, region, typ, num):

req_ins_num = []
[ req_ins_num.append((len(ins['instance_id_list']), r)) for r, ins in info.items() if ins['type'] == typ and ins['region'] == region ]

Comment on lines -270 to +266

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function kill_spot_instances_by_num refactored with the following changes:

logging.info(f'Will kill {num} {typ} instances')

cancel_req_ids = []
Expand All @@ -280,15 +276,15 @@ def kill_spot_instances_by_num(name, region, typ, num):
cancel_req_ids.append(req)

logging.info(f'Kill {total_size} {typ} instances')
if len(cancel_req_ids) > 0:

if cancel_req_ids:
cancel_spot_instances(name, cancel_req_ids)

def cancel_spot_instances(name, request_ids):
info = aws_accessor.get_cluster(name)['info']
for i in request_ids:
if i not in info:
logging.info('Request ID : {} not found'.format(i))
logging.info(f'Request ID : {i} not found')
Comment on lines -291 to +287

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function cancel_spot_instances refactored with the following changes:

continue
instance_id_list = info[i]['instance_id_list']
region = info[i]['region']
Expand All @@ -298,47 +294,47 @@ def cancel_spot_instances(name, request_ids):
TerminateInstances=True
)
if 'SuccessfulFleetRequests' in res:
logging.info('Successful cancel spot fleet request {}'.format(i))
logging.info(f'Successful cancel spot fleet request {i}')
instance_accessor.del_instance(name, [ i.__dict__ for i in utils.get_ins_from_ids(region, instance_id_list)])
aws_accessor.del_request(name, i)

def cancel_all_instances(name):
requests = aws_accessor.get_requests(name)
if requests:
if requests := aws_accessor.get_requests(name):
Comment on lines -306 to +302

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function cancel_all_instances refactored with the following changes:

cancel_spot_instances(name, requests)


def get_client(region=DEFAULT_REGION):
return boto3.client('ec2', region_name=region, **CREDENTIALS)

def _get_request_config(params):
base = {
return {
'TargetCapacity': params['targetCapacity'],
'TerminateInstancesWithExpiration': True,
'ValidFrom': datetime(2018, 1, 1),
'ValidUntil': datetime(2019, 1, 1),
'IamFleetRole': 'arn:aws:iam::906727922743:role/aws-ec2-spot-fleet-role',
'LaunchSpecifications': [{
'ImageId': params['imageId'],
'KeyName': KEYS[params['region']],
'InstanceType': params['instanceType'],
'BlockDeviceMappings': [{
'VirtualName': 'Root',
'DeviceName': '/dev/sda1',
'Ebs': {
'VolumeSize': 75,
'VolumeType': 'gp2',
'DeleteOnTermination': True
}
}],
'Monitoring': {
'Enabled': False
'LaunchSpecifications': [
{
'ImageId': params['imageId'],
'KeyName': KEYS[params['region']],
'InstanceType': params['instanceType'],
'BlockDeviceMappings': [
{
'VirtualName': 'Root',
'DeviceName': '/dev/sda1',
'Ebs': {
'VolumeSize': 75,
'VolumeType': 'gp2',
'DeleteOnTermination': True,
},
}
],
'Monitoring': {'Enabled': False},
}
}],
],
'AllocationStrategy': 'lowestPrice',
'Type': 'maintain'
'Type': 'maintain',
}
return base
Comment on lines -315 to -341

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function _get_request_config refactored with the following changes:


def _send_request(client, params):
"""
Expand All @@ -351,7 +347,7 @@ def _send_request(client, params):
res = client.request_spot_fleet(
SpotFleetRequestConfig=_get_request_config(params)
)
logging.info('Created spot fleet request {}'.format(res["SpotFleetRequestId"]))
logging.info(f'Created spot fleet request {res["SpotFleetRequestId"]}')
Comment on lines -354 to +350

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function _send_request refactored with the following changes:

return res['SpotFleetRequestId']

def _wait_initialized(client, instance_id_list):
Expand All @@ -364,7 +360,10 @@ def _wait_initialized(client, instance_id_list):
if len(res['InstanceStatuses']) == 0:
time.sleep(10)
continue
if all([ s['InstanceStatus']['Status'] == 'ok' for s in res['InstanceStatuses'] ]):
if all(
s['InstanceStatus']['Status'] == 'ok'
for s in res['InstanceStatuses']
):
Comment on lines -367 to +366

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function _wait_initialized refactored with the following changes:

logging.info('Instances are initialized now.')
return
time.sleep(10)
Expand All @@ -381,8 +380,7 @@ def _wait_active(client, request_id, max_num):
res = client.describe_spot_fleet_instances(SpotFleetRequestId=request_id)
if len(res['ActiveInstances']) == max_num:
logging.info('Instances are active now.')
instance_id_list = [ i['InstanceId'] for i in res['ActiveInstances'] ]
return instance_id_list
return [ i['InstanceId'] for i in res['ActiveInstances'] ]
Comment on lines -384 to +383

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function _wait_active refactored with the following changes:

time.sleep(10)

def _set_security_group(client, instance_id_list, security_groups):
Expand Down
12 changes: 4 additions & 8 deletions modules/data_accessor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,7 @@ def save_prizes(self, prizes):
)

def get_prize(self, region):
record = self.collection.find_one({'region' : region})
if record:
if record := self.collection.find_one({'region': region}):

@sourcery-ai sourcery-ai Bot Jul 5, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Function PrizeAccessor.get_prize refactored with the following changes:

return record['sizes']

class AWSAccessor(_BaseAccessor):
Expand All @@ -40,8 +39,7 @@ def save_cluster(self, name, info):
self.collection.update({'name' : name}, doc, upsert=True)

def get_by_region_typ(self, name, region, typ):
record = self.collection.find_one({'name' : name})
if record:
if record := self.collection.find_one({'name': name}):
for r, info in record['info'].items():
if info['region'] == region and info['type'] == typ:
return (r, info['instance_id_list'])
Expand All @@ -61,8 +59,7 @@ def del_requests(self, name, ids):
[self.del_request(name, id) for id in ids]

def get_requests(self, name):
record = self.collection.find_one({'name' : name})
if record:
if record := self.collection.find_one({'name': name}):
return record['info'].keys()

class InstanceAccessor(_BaseAccessor):
Expand All @@ -75,8 +72,7 @@ def update_instances(self, name, instance_list):
self.collection.update({'name' : name}, doc, upsert=True)

def get_instances(self, name):
record = self.collection.find_one({'name' : name})
if record:
if record := self.collection.find_one({'name': name}):
return record['instances']

def get_all_instances(self):
Expand Down
Loading