Evaluasi Solusi Cloud

import http from 'k6/http'; import { check, sleep } from 'k6'; import { Rate, Trend, Counter } from 'k6/metrics'; // Custom metrics const failureRate = new Rate('failed_requests'); const requestDuration = new Trend('request_duration'); const totalRequests = new Counter('total_requests'); export const options = { stages: [ { duration: '2m', target: 100 }, // Ramp-up { duration: '5m', target: 100 }, // Stable load { duration: '2m', target: 200 }, // Spike { duration: '3m', target: 200 }, // High load { duration: '2m', target: 0 }, // Ramp-down ], thresholds: { http_req_duration: ['p(95)<500'], // 95% requests under 500ms failed_requests: ['rate<0.05'], // Less than 5% failures http_reqs: ['count>1000'], // More than 1000 requests }, }; export default function () { const urls = [ 'https://api.example.com/products', 'https://api.example.com/orders', 'https://api.example.com/users', ]; const url = urls[Math.floor(Math.random() * urls.length)]; const params = { headers: { 'Authorization': 'Bearer ' + __ENV.API_TOKEN, 'Content-Type': 'application/json', }, tags: { name: 'api_request' }, }; totalRequests.add(1); const response = http.get(url, params); const success = check(response, { 'status is 200': (r) => r.status === 200, 'response time < 500ms': (r) => r.timings.duration < 500, 'has response body': (r) => r.body.length > 0, }); failureRate.add(!success); requestDuration.add(response.timings.duration); sleep(1); }

import boto3 import json import time from datetime import datetime, timedelta import pandas as pd import matplotlib.pyplot as plt class PerformanceMonitor: def __init__(self): self.cloudwatch = boto3.client('cloudwatch') self.ec2 = boto3.client('ec2') def get_performance_metrics(self, instance_id, hours=24): """Collect comprehensive performance metrics""" end_time = datetime.utcnow() start_time = end_time - timedelta(hours=hours) metrics = { 'cpu_utilization': self.get_cpu_metrics(instance_id, start_time, end_time), 'memory_usage': self.get_memory_metrics(instance_id, start_time, end_time), 'network_io': self.get_network_metrics(instance_id, start_time, end_time), 'disk_io': self.get_disk_metrics(instance_id, start_time, end_time), } return metrics def get_cpu_metrics(self, instance_id, start_time, end_time): response = self.cloudwatch.get_metric_statistics( Namespace='AWS/EC2', MetricName='CPUUtilization', Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}], StartTime=start_time, EndTime=end_time, Period=300, Statistics=['Average', 'Maximum'] ) return response['Datapoints'] def analyze_performance_trends(self, metrics): """Analyze performance trends and identify issues""" analysis = { 'cpu_bottlenecks': self.analyze_cpu_bottlenecks(metrics['cpu_utilization']), 'memory_issues': self.analyze_memory_issues(metrics['memory_usage']), 'network_bottlenecks': self.analyze_network_issues(metrics['network_io']), 'recommendations': [] } # Generate recommendations if analysis['cpu_bottlenecks']['high_usage_periods'] > 0: analysis['recommendations'].append( "Consider upgrading instance type due to consistent high CPU usage" ) return analysis def generate_performance_report(self, instance_id): """Generate comprehensive performance report""" metrics = self.get_performance_metrics(instance_id) analysis = self.analyze_performance_trends(metrics) report = { 'instance_id': instance_id, 'evaluation_period': '24 hours', 'metrics_summary': { 'avg_cpu_utilization': self.calculate_average(metrics['cpu_utilization']), 'max_cpu_utilization': self.calculate_maximum(metrics['cpu_utilization']), 'performance_score': self.calculate_performance_score(analysis) }, 'analysis': analysis, 'timestamp': datetime.utcnow().isoformat() } return report # Usage monitor = PerformanceMonitor() report = monitor.generate_performance_report('i-1234567890abcdef0') print(json.dumps(report, indent=2))

import boto3 import time from datetime import datetime import threading class AutoScalingEvaluator: def __init__(self): self.autoscaling = boto3.client('autoscaling') self.cloudwatch = boto3.client('cloudwatch') self.ec2 = boto3.client('ec2') def trigger_scaling_event(self, asg_name, target_cpu=80): """Trigger auto-scaling by simulating high CPU load""" print(f"Triggering scaling event for ASG: {asg_name}") # Get current instances response = self.autoscaling.describe_auto_scaling_groups( AutoScalingGroupNames=[asg_name] ) initial_capacity = len(response['AutoScalingGroups'][0]['Instances']) print(f"Initial capacity: {initial_capacity} instances") # Simulate CPU load to trigger scaling self.simulate_cpu_load(asg_name, target_cpu) # Monitor scaling activities self.monitor_scaling_activities(asg_name, initial_capacity) def simulate_cpu_load(self, asg_name, target_cpu): """Simulate CPU load using CloudWatch metrics""" # This would typically be done by actually generating load # For simulation, we'll put custom metrics instances = self.get_asg_instances(asg_name) for instance_id in instances: self.cloudwatch.put_metric_data( Namespace='AWS/EC2', MetricData=[ { 'MetricName': 'CPUUtilization', 'Dimensions': [ { 'Name': 'InstanceId', 'Value': instance_id }, ], 'Timestamp': datetime.utcnow(), 'Value': target_cpu, 'Unit': 'Percent' }, ] ) print(f"Simulated {target_cpu}% CPU load on {len(instances)} instances") def monitor_scaling_activities(self, asg_name, initial_capacity, timeout=600): """Monitor scaling activities and measure performance""" start_time = time.time() scaling_detected = False print("Monitoring scaling activities...") while time.time() - start_time < timeout: response = self.autoscaling.describe_scaling_activities( AutoScalingGroupName=asg_name, MaxRecords=10 ) current_activities = response['Activities'] # Check for scaling activities for activity in current_activities: if activity['StatusCode'] == 'InProgress': scaling_detected = True print(f"Scaling activity detected: {activity['Description']}") # Measure scaling time scaling_start = activity['StartTime'] scaling_duration = datetime.utcnow() - scaling_start print(f"Scaling duration: {scaling_duration}") # Check current capacity response = self.autoscaling.describe_auto_scaling_groups( AutoScalingGroupNames=[asg_name] ) current_capacity = len(response['AutoScalingGroups'][0]['Instances']) if current_capacity > initial_capacity and scaling_detected: print(f"Scaling completed! New capacity: {current_capacity} instances") return { 'success': True, 'scaling_time': time.time() - start_time, 'new_capacity': current_capacity, 'instances_added': current_capacity - initial_capacity } time.sleep(10) return {'success': False, 'error': 'Scaling timeout'} # Evaluation execution evaluator = AutoScalingEvaluator() result = evaluator.trigger_scaling_event('my-auto-scaling-group', 85) print("Scaling evaluation result:", result)

import boto3 import json from datetime import datetime from typing import Dict, List class SecurityScanner: def __init__(self): self.ec2 = boto3.client('ec2') self.iam = boto3.client('iam') self.s3 = boto3.client('s3') self.securityhub = boto3.client('securityhub') def comprehensive_scan(self) -> Dict: """Perform comprehensive security scan""" scan_results = { 'timestamp': datetime.utcnow().isoformat(), 'findings': [], 'risk_level': 'LOW', 'recommendations': [] } # Run various security checks scan_results['findings'].extend(self.check_network_security()) scan_results['findings'].extend(self.check_iam_security()) scan_results['findings'].extend(self.check_data_protection()) scan_results['findings'].extend(self.check_monitoring()) # Calculate overall risk level scan_results['risk_level'] = self.calculate_risk_level(scan_results['findings']) # Generate recommendations scan_results['recommendations'] = self.generate_recommendations(scan_results['findings']) return scan_results def check_network_security(self) -> List[Dict]: """Check network security configurations""" findings = [] # Check security groups response = self.ec2.describe_security_groups() for sg in response['SecurityGroups']: for permission in sg.get('IpPermissions', []): # Check for overly permissive rules for ip_range in permission.get('IpRanges', []): if ip_range['CidrIp'] == '0.0.0.0/0': findings.append({ 'severity': 'HIGH', 'category': 'NETWORK_SECURITY', 'resource': sg['GroupId'], 'description': f"Security group {sg['GroupId']} allows inbound traffic from anywhere", 'recommendation': 'Restrict inbound traffic to specific IP ranges' }) return findings def check_iam_security(self) -> List[Dict]: """Check IAM security configurations""" findings = [] # Check IAM policies response = self.iam.list_policies(Scope='Local') for policy in response['Policies']: policy_version = self.iam.get_policy_version( PolicyArn=policy['Arn'], VersionId=policy['DefaultVersionId'] ) # Analyze policy document for excessive permissions if self.has_admin_permissions(policy_version['PolicyVersion']['Document']): findings.append({ 'severity': 'HIGH', 'category': 'IAM_SECURITY', 'resource': policy['PolicyName'], 'description': f"IAM policy {policy['PolicyName']} has administrative permissions", 'recommendation': 'Apply principle of least privilege' }) return findings def generate_security_report(self, scan_results: Dict) -> str: """Generate human-readable security report""" report = f""" # Cloud Security Assessment Report **Generated:** {scan_results['timestamp']} **Overall Risk Level:** {scan_results['risk_level']} ## Executive Summary - Total Findings: {len(scan_results['findings'])} - High Severity: {len([f for f in scan_results['findings'] if f['severity'] == 'HIGH'])} - Recommendations: {len(scan_results['recommendations'])} ## Critical Findings """ high_findings = [f for f in scan_results['findings'] if f['severity'] == 'HIGH'] for finding in high_findings[:5]: # Top 5 critical findings report += f"\n### {finding['category']}\n" report += f"- **Resource:** {finding['resource']}\n" report += f"- **Description:** {finding['description']}\n" report += f"- **Recommendation:** {finding['recommendation']}\n" report += "\n## Recommendations\n" for rec in scan_results['recommendations'][:5]: # Top 5 recommendations report += f"- {rec}\n" return report # Execute security scan scanner = SecurityScanner() results = scanner.comprehensive_scan() report = scanner.generate_security_report(results) print(report)