클라우드 개발자 앙몬드

쿠버네티스 COP를 토대로, CICD를 운영하다보면 간혹 이런 현상이 나타날 때가 있습니댜

 - 예를 들면.... CICD 중도 파이프라인 강제종료 (웹 통신에서 Broken Pipe 격)

앙몬드는 젠킨스를 이용한 CI & CD를 주로 사용했는데, 젠킨스 잡이 끊어지면, CD 과정이 꺠지기에, 배포 정합성이 어긋나게 되는 케이스가 발생합니댜

 - ex) 블루 live / 그린 live / 쿠버프록시 스위치 미실행

          따라서, 블루 & 그린 양쪽 사이드의 파드가 동시에 실행되고 있으나, 실제 요청은 블루 사이드로만 리퀘스트가 라우트

          이에 대한 검증으로, 좀비 파드를, 쿠버프록시 기준으로 색출하는 작업이 필요

간단하게 파이썬으로 작성해본, 쿠버프록시 기준 좀비파드 색출 스크립트입니댜

import boto3
import time
import os
import decimal
import sys
import subprocess

from decimal import Decimal
from datetime import datetime, timedelta
from collections import deque

"""
쿠버프록시 to 파드 매핑 출력 스크립트
 - 블루/그린 파드 동시 live시, 한쪽 사이드 저격용

 @since 2020-07-27
 @author angmond

 [2020-07-27] 최초생성 (angmond)
"""

print("[K8S 시스템 스크립트 - 쿠버프록시 to 파드 매핑 출력 스크립트 ")
print("\n\n")

"""(1) 네임스페이스 아규먼드 분리 """
namespace = sys.argv[1]
print("입력 네임스페이스 : " + namespace)

"""(2) 쿠버 네임스페이스에 속하는 전체 파드 출력"""
all_pods = subprocess.check_output("kubectl get pod --namespace " + namespace, shell=True, universal_newlines=True)
list_pods = str(all_pods).split("\n")
list_pods_name_and_status = []
for idx, value in enumerate(list_pods) :
    if idx is 0:
        continue

    sublist = " ".join(value.split())
    if sublist is "":
        continue

    pod_name = sublist.split(" ")[0]
    pod_status = sublist.split(" ")[2]
    list_pods_name_and_status.append(pod_name + " / " + pod_status)

"""(3) 현재 서비스파드 (쿠버프록시) 전체 조회 """
all_kube_proxy = subprocess.check_output("kubectl describe svc --namespace " + namespace + " | grep Selector", shell=True, universal_newlines=True)
list_raw_kube_proxy = str(all_kube_proxy).split("\n")
list_kube_proxy = []
for idx, value in enumerate(list_raw_kube_proxy):

    sublist = " ".join(value.split())
    if sublist is "":
        continue

    service_name_and_route_pod_info = sublist.split(" ")[1]
    service_name_and_route_pod_info = service_name_and_route_pod_info.replace("name=", "")
    service_name_and_route_pod_info = service_name_and_route_pod_info.replace(",version=", "-")
    list_kube_proxy.append(service_name_and_route_pod_info)

"""(4) 쿠버 프록시 리스트를 기준으로, 매핑되는 동일네임 파드를 출력한다 """
list_kube_proxy.sort()
list_complete_route_proxy = []
print("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
print("++++++++++++++++Kube-proxy -> Pod Mapping Info++++++++++++++++")
for proxy_info in list_kube_proxy:
    list_abnormal_pod = []
    print("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
    for pod_info in list_pods_name_and_status:
        if proxy_info in pod_info:
            kube_proxy_pod_set = proxy_info + " -> " + pod_info 
            list_complete_route_proxy.append(kube_proxy_pod_set)
            print("+++ " + kube_proxy_pod_set)
    print("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")

print("\n\n")
print("++++++++++++++++++++++++[비정상 파드 탐색]++++++++++++++++++++++++")
"""(5) 쿠버프록시와 매핑되지 않은, 비정상 파드를 출력한다. """
for pod_info in list_pods_name_and_status:
    is_pod_not_mapped = True 
    for mapped_info in list_complete_route_proxy:
        if pod_info in mapped_info :
            is_pod_not_mapped = False
            break 
    if is_pod_not_mapped :
        print("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
        print(f"+++ {pod_info} ")
        print("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++")
print("++++++++++++++++++++++++[비정상 파드 종료]++++++++++++++++++++++++")

계속 업무 정리를 하다가, 나오게 되는 중간중간 코드 한줄입니다.

AWS 클라우드 네이티브 환경에서, 파이썬 런타임을 이용해,

  (1) ES 접근을 위한 토큰 발급

  (2) ES커넥터 생성

  (3) ES인덱싱

에 대한 과정을 간략히 요약한 코드입니다.

from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection
import boto3 

#... 선행코드 중략

# boto3를 이용해, 본 파이썬이 기동되는 클라우드 베이스 호스트에서, ES에 접근하기 위한, 토큰을 획득합니다.
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(
    credentials.access_key, 
    credentials.secret_key, 
    "ap-northeast-2", 
    "es", 
        session_token=credentials.token
)

# AWS 엘라스틱서치 매니지드 클러스터가 생성되어 있다고 가정하고, 그에 대한 엔드포인트를 기입합니다.
# 엘라스틱서치 커넥터를 생성하기 위한 포맷은 하기와 같습니다.
host = 'vpc-es-cloud-developer-angmond.ap-northeast-2.es.amazonaws.com'
es_connector = Elasticsearch(
    hosts = [{'host': host, 'port': 443}],
    http_auth = awsauth,
    use_ssl = True,
    verify_certs = True,
    connection_class = RequestsHttpConnection
)

# 엘라스틱서치 인덱스명을 위해, 시계열 패턴 yyyy.MM
today = datetime.today()
es_index = "k8sreport-" + str(today.year) + '.' + str(today.month)

# 무언가의_뒤죽박죽_데이터 콜렉션 루프를 돌며, 엘라스틱서치에 데이터를 인덱싱합니다.
for 아이템 in 무언가의_뒤죽박죽_데이터:
    es_connector.index(index=es_index, body=아이템)

근 두달가까이 소식이 없다가, 그동안 쌓인 내용들을 부랴부랴 올리고 있습니다.

사실 파이썬3에 대한 이용능력을 좀 높일 필요가 있어서, 해커랭크에서 좀 살다 왔습니다.

신기한게 자격증도 주어지더라구요... (어느정도 문제 풀어보니)

한번 들어가셔서 확인해보시기 바랍니다 :) 

www.hackerrank.com

자 무튼, 한동안 소식이 없기도 했고, 그 소식 없는 기간동안 잠깐 과제를 진행하기도 했습니다.

로컬스택을 이용해, 로컬 환경에서의 AWS 아키텍처링을 테스팅해볼 수 있는 도구입니다.

로컬스택에 대한 설명은 조금 후일담으로 미루어둘게요. (클라우드 개발자이기에 코드를 집중적으로 판다 실시)

(아참, 환경마다 다를 순 있겠지만, 파이썬3를 이용해 로컬스택 설치하지 마시구, 도커 & 도커컴포즈파일 이용해서 설치하세요. 파이썬3 이용해서 설치할 때, 디펜던시 계속 깨지는 오류가 발생합니다. 트러블 슈팅 귀찮으니 도커/컴포즈 로 가는 방법 추천드립니다.)

* 로컬스택 main

https://hub.docker.com/r/localstack/localstack/

* 로컬스택 설치 가이드

https://medium.com/better-programming/dont-be-intimidated-learn-how-to-run-aws-on-your-local-machine-with-localstack-2f3448462254

과제로 주어진, 구성해야할 아키텍처는 다음과 같습니다.

로컬스택 환경이 모두 구성되었고, 로컬스택 컨테이너가 돌아가고 있다고 가정하겠습니다.

이제 저 모든 구성요소를 의존성 순서대로 주입하여, AWS boto3 & python3 스크립트를 돌리게 할 겁니다.

한편으로는 테라폼 IAC로도 되긴 하지만, 충분히 SDK로도 됩니다..!!!!! 라는 것을 보여주고 싶었어요 ㄷㄷㄷㄷ.... (그래봤자 100% 치환은 되진 않겠지만..)

import json
import boto3
import subprocess

"""
로컬스택 AWS 클라우드 인프라 메이커 (Python3 Script with boto3)

@desc 
	POC 수행을 위한, AWS 서비스 구성요소
	본 파이썬 스크립트 실행 시, 하기 구성요소 생성
	
	IAM : 루트유저 급 마스터 Role 
	IGW : 신규 인터넷 게이트웨이
	VPC : 신규 VPC (10.0.0.0/16)
	Subnets Of VPC : 신규 VPC의 서브넷
	RouteTable (RT) : 생성된 서브넷을 위한 라우트테이블
	Route of RT : 라우트 테이블의 네트워크 경로
	Security Group : 모든 대역 / 모든 포트 Open (AnyOpen) (In/Out 포함)
	EC2 Key Pair : EC2 머신을 위한 키페어
	EC2 Instance Profile : EC2 머신이 사용할 IAM 프로필
	EC2 Instance : Nginx가 기동될 EC2 머신
	Kinesis : EC2 머신이 로그를 전송할 키네시스 데이터스트림
	Lambda : 키네시스 데이터스트림의 이벤트 트리거를 받아 처리할 람다
	Lambda Trigger : 키네시스 DS와 람다 이벤트 매핑
	S3 : 람다가 정제한 로그를 쌓을 버킷
@author 클라우드 개발자 앙몬드
@since 2020-09-17
"""

print("[System] 로컬스택 AWS 클라우드 인프라 메이커 시작")
print("=" * 30)

PROFILE_NAME = "localstack"			# aws API와 통신하기 위한 크리덴셜 프로필
REGION_NAME = "ap-northeast-2"			# 로컬스턱 리전 고정 (서울)
ENDPOINT_URL = "http://localhost:4566/"		# 로컬스택 Fake AWS API 엔드포인트

session = boto3.session.Session(profile_name=PROFILE_NAME, region_name=REGION_NAME)	# boto3 세션 팩토리
iam_client = session.client("iam", endpoint_url=ENDPOINT_URL)		# iam 클라이언트 
ec2_resource = session.resource("ec2", endpoint_url=ENDPOINT_URL)		# ec2 리소스
ec2_client = session.client("ec2", endpoint_url=ENDPOINT_URL)		# ec2 클라이언트
kinesis_client = session.client('kinesis', endpoint_url=ENDPOINT_URL)	# kinesis 클라이언트
lambda_client = session.client('lambda', endpoint_url=ENDPOINT_URL)	# lambda 클라이언트
s3_client = session.client("s3", endpoint_url=ENDPOINT_URL)			# s3 클라이언트

LOCALSTACK_LOG_S3_BUCKET_NAME = "localstack-fake-log-bucket"		# s3 버킷 이름
LOCALSTACK_KINESIS_DS_NAME = "localstack-fake-kinesis-ds-for-log"		# 키네시스 이름
LOCALSTACK_LAMBDA_NAME = "localstack-fake-lambda-for-log-processing"	# 람다함수 이름
print("[System] 기본 환경변수 설정 완료")

"""
루트유저 급 마스터 Role 생성
"""
# IAM Policy 생성
managed_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Sid": "VisualEditor0",
        "Effect": "Allow",
        "Action": "*",
        "Resource": "*"}]}
response_create_policy = iam_client.create_policy(
    PolicyName='localstack-fake-master-policy',
    PolicyDocument=json.dumps(managed_policy))
arn_fake_master_policy = response_create_policy["Policy"]["Arn"]

# IAM Role 생성
trust_entity_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {
            "Service": [
                "lambda.amazonaws.com",
                "s3.amazonaws.com",
                "kinesis.amazonaws.com",
                "ec2.amazonaws.com"
            ]
        },
        "Action": "sts:AssumeRole"}]}
response = iam_client.create_role(
    Path="/",
    RoleName="localstack-fake-master-role",
    AssumeRolePolicyDocument=json.dumps(trust_entity_policy),
    Description="localstack-fake-master-role",
    MaxSessionDuration=3600)
iam_role_arn = response["Role"]["Arn"]

# IAM Role에 Policy 탑재
# arn_fake_master_policy 은, create_policy 의 리턴값으로 획득 가능
iam_client.attach_role_policy(
    PolicyArn=arn_fake_master_policy,
    RoleName='localstack-fake-master-role')
print("[System] AWS 계정 루트유저급 IAM Role 생성 완료 - 루트유저급 Role 사용 가능 엔티티 : EC2 / Lambda / S3 / Kinesis")

"""
인터넷 게이트웨이 / VPC 생성 & IGW 탑재
"""
igw = ec2_resource.create_internet_gateway()
vpc = ec2_resource.create_vpc(CidrBlock='10.0.0.0/16')
ec2_client.modify_vpc_attribute(VpcId=vpc.id, EnableDnsSupport={'Value': True})
ec2_client.modify_vpc_attribute(VpcId=vpc.id, EnableDnsHostnames={'Value': True})
vpc.attach_internet_gateway(InternetGatewayId=igw.id)
print("[System] IGW/VPC 생성 & 신규 생성 VPC에 IGW 링크 완료")

"""
VPC의 서브넷 생성 (AZ 이중화를 위해 2개로 분할 생성), RT 생성 & 네트워크 라우터 탑재
	-> nginx 인스턴스는 이중화되지 않으므로, 서브넷은 이중화하였으며, 실제 라우터는 한 곳만 설정
"""
routetable = vpc.create_route_table()
route_public = routetable.create_route(DestinationCidrBlock='0.0.0.0/0', GatewayId=igw.id)
subnet_a = ec2_resource.create_subnet(CidrBlock='10.0.0.0/24', VpcId=vpc.id, AvailabilityZone="ap-northeast-2a")
subnet_c = ec2_resource.create_subnet(CidrBlock='10.0.1.0/24', VpcId=vpc.id, AvailabilityZone="ap-northeast-2c")
routetable.associate_with_subnet(SubnetId=subnet_a.id)
print("[System] Route Table 및 요청 네트워크 라우팅 설정 완료")
print("[System] VPC IP대역 중, 이중화를 위한 subnet C클래스 분할 & AZ-A 서브넷에 라우트 탑재 완료")

"""
인/아웃바운드 전 대역 전 포트 오픈 Security Group  생성
"""
security_group = ec2_resource.create_security_group(GroupName='sg-localstack', Description='sg-localstack', VpcId=vpc.id)
security_group.authorize_ingress(
    IpPermissions=[
        {
            'FromPort': 0,
            'IpProtocol': 'tcp',
            'IpRanges': [
                {
                    'CidrIp': '0.0.0.0/0',
                    'Description': 'all_any_open'
                },
            ],
            'ToPort': 65535
        },
        {
		'FromPort': 0,
		'ToPort': 65535,
		'IpProtocol': 'udp',
		'IpRanges': [
			{
				'CidrIp': '0.0.0.0/0',
				'Description': 'all_any_open'
			},
		], 
    }])
print("[System] 시큐리티그룹 네트워크 애니오픈 생성 완료")

"""
EC2에 접근하기 위한 Key Pair 생성
"""
LOCAL_STACK_FAKE_KEYPAIR_NAME = "fake-keypair"
fake_keypair = open('fake-keypair.pem', 'w')
key_pair_response = ec2_resource.create_key_pair(KeyName=LOCAL_STACK_FAKE_KEYPAIR_NAME)
KeyPairOut = str(key_pair_response.key_material)
fake_keypair.write(KeyPairOut)
fake_keypair.close()
print("[System] EC2 머신 접근을 위한 키페어 생성 완료 > ./fake-keypair.pem 확인")


"""
nginx 웹서버를 위한 EC2 인스턴스 생성 
"""
instance_profile = iam_client.create_instance_profile(InstanceProfileName='ec2-nginx-instance-profile')
arn_of_instance_profile = instance_profile["InstanceProfile"]["Arn"]
response = iam_client.add_role_to_instance_profile(
    InstanceProfileName='ec2-nginx-instance-profile',
    RoleName='localstack-fake-master-role'
)
print("[System] EC2 머신을 위한 IAM 인스턴스 프로필 생성 완료")

instances = ec2_resource.create_instances(
    ImageId='ami-06c2d58d267b830df',
    InstanceType='t2.micro',
    KeyName=LOCAL_STACK_FAKE_KEYPAIR_NAME,
    MinCount=1,
    MaxCount=1,
    Placement={'AvailabilityZone': 'ap-northeast-2a' },
    NetworkInterfaces=[{
        'SubnetId': subnet_a.id,
        'DeviceIndex': 0,
        'AssociatePublicIpAddress': True,
        'Groups': [security_group.group_id]  }],
    IamInstanceProfile={
        'Arn': arn_of_instance_profile,
        'Name': 'ec2-nginx-instance-profile' }
)
print("[System] EC2 머신 생성 완료")

"""
키네시스 생성 시, 요청에 대한 응답은 200 정상으로 리턴되지만, 실제 생성되지 않는 이슈가 발생
"""
#subprocess.run('aws kinesis create-stream --stream-name "localstack-fake-kinesis-ds-for-log" --shard-count 1 --profile localstack --endpoint http://localhost:4566', shell=True, check=True)
response = kinesis_client.create_stream(StreamName=LOCALSTACK_KINESIS_DS_NAME, ShardCount=1)
#response2 = kinesis_client.list_streams( Limit=100 )
print(f"[DEBUG] Kinesis Creation Response : {response}")
print(f"[System] 키네시스 DS 생성 완료 (응답 Success / 실제 구현은 안되는 버그 존재)")
print("[Warn] 로컬스택 키네시스 ARN : arn:aws:kinesis:ap-northeast-2:000000000000:stream/localstack-kinesis-fake-ds")



"""
람다 생성
"""
response_lambda = lambda_client.create_function(
	FunctionName=LOCALSTACK_LAMBDA_NAME,
	Runtime='python3.8',
	Role=iam_role_arn,
	Handler = 'lambda_function.lambda_handler',
	Code={'ZipFile': open('./localstack-logs-process-fake-lambda.zip', 'rb').read()},
	Description='localstack-lambda',
	Timeout=900,
	MemorySize=1024,
	Publish=True,
	VpcConfig={
		   'SubnetIds': [subnet_a.id, subnet_c.id ],
		   'SecurityGroupIds': [security_group.group_id ] } )
print("[System] 람다 생성 완료")

response_lambda_trigger = lambda_client.create_event_source_mapping(
    EventSourceArn='arn:aws:kinesis:ap-northeast-2:000000000000:stream/localstack-kinesis-fake-ds',
    FunctionName=LOCALSTACK_LAMBDA_NAME,
    Enabled=True,
    BatchSize=100,
    MaximumBatchingWindowInSeconds=60,
    ParallelizationFactor=1,
    MaximumRecordAgeInSeconds= -1,
    BisectBatchOnFunctionError=False,
    MaximumRetryAttempts=0,
)
print("[System] 람다 이벤트 트리거 매핑 완료 <- Kinesis Fake Stream of Bugs")

"""
s3 버킷 생성 
"""
response_s3_create = s3_client.create_bucket(
    ACL='private',
    Bucket=LOCALSTACK_LOG_S3_BUCKET_NAME,
    CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-2' },
)
print(f"[System] S3 버킷 생성 완료")
print("=" * 30)
print("[System] 로컬스택 AWS 클라우드 인프라 메이커 종료")

한동안 파이썬3 알고리즘 / 코딩 공부좀 한두달 빠져있다가 돌아왔습니다.

그리고 얼마전에는 로컬스택으로 조금 재밌게 놀아보기도 했는데, 로컬스택 활용 미션 중, 람다 함수도 만드는게 있더라구요.

이건 로컬스택을 이용한 아키텍처링 과제로, 그 중 키네시스 -> 람다 이벤트 드라이븐 구간에서 람다 백엔드 함수입니다.

* 진짜 과제는, 로컬스택을 이용하여 저 모든 아키텍처를 구성하고 POC 검증을 하라는 내용이었습니다.

* 보너스 미션으로, 테라폼 (iac)를 이용하여 이 아키텍처를 원콤내는 것이 있는데, 저는 boto3 sdk를 이용해 코드 런타임으로 한방에 구현을 해보겠습니다. (다음 글..!)

devloper-angmond.tistory.com/30

import json
import base64
import boto3
from datetime import datetime

def lambda_handler(event, context):
    print(f"로그를 정제합니다.")
    list_raw_record = event["Records"]
    list_converted_logs = []
    s3버킷의_이름 = "cloud-developer-angmonddddddd"
    
    # 로그 정제
    # KDS는 샤드 안의 데이터를 base64 인코딩하므로, 같은 방식으로 디코딩해야
    # 사람의 인식이 쉬운 데이터로 변환됨
    for kinesis_data in list_raw_record :
        raw_data_base64_encoded = kinesis_data["kinesis"]["data"]
        raw_data_decoded = base64.b64decode(raw_data_base64_encoded)
        list_converted_logs.append(raw_data_decoded)
    
    s3_client = boto3.client('s3')
    # 오브젝트 put 시, S3 또한 파티션 단위이므로, 1분단위 로그가 키네시스 입력 & 람다 트리거링이므로, yyyy-mm-dd-hh-mm 단위로 S3 오브젝트 파일명을 작성
    s3_client.put_object(Body=list_converted_logs, Bucket=s3버킷의_이름, Key="/nginx_logs/" + datetime.now().strftime("%Y-%m-%d %H:%M") )
        
    print(f"로그 정제작업이 끝났습니다.")

# OrderedDict
# 슈퍼마켓에, N개 아이템 & 가격 리스트가 있다
# 이 n개를, 손님들이 특정 날짜에 샀다
# 내 업무는, 각 아이템명과 가격을 순서대로 출력하는 것이다
#
exit(0)
from collections import OrderedDict
if __name__ == "__main__" :
    전체장부개수 = int(input())
    장부 = OrderedDict()
    for i in range(전체장부개수) :
        입력값 = input().split()
        아이템명 = " ".join(입력값[:-1])
        가격 = int(입력값[-1])
        try :
            장부[아이템명] += 가격
        except Exception as e :
            장부[아이템명] = 가격

    for 아이템, 매출 in 장부.items():
        print(f"{아이템} {매출}")

# itertools combination의 심화버전. 자신마저 끼고 조합찾기
exit(0)
from itertools import combinations_with_replacement
if __name__ == "__main__" :
    입력값 = input().split()
    피봇 = int(입력값[1])
    정렬위한_리스트 = list(입력값[0])
    정렬위한_리스트.sort()
    단어 = "".join(정렬위한_리스트)
    결과 = list(combinations_with_replacement(단어,피봇))
    for i in 결과 :
        print("".join(i))
        
# itertools.combinations(콜렉션, r)
# 콜렉션들의 요소를 모두 조합하여 콜렉션 타입의 리스트 생성
exit(0)
from itertools import combinations
if __name__ == "__main__" :
    입력값 = input().split()
    임시 = list(입력값[0])
    임시.sort()
    스트링 = "".join(임시)
    피봇 = int(입력값[1])
    조합결과_리스트 = []
    for i in range(피봇) :
        임시_리스트 = list(combinations( 스트링, (i+1) ))
        조합결과_리스트.extend(임시_리스트)
    for e in 조합결과_리스트 :
        print("".join(e))
        
# 퍼뮤테이션.
# 콜렉션을 통해, 가능한 모든 조합 서브콜렉션을 탄생시키는 라이브러리
exit(0)
from itertools import permutations
def 퍼뮤테이션(string, k) :
    퍼뮤테이션_결과리스트 = list(permutations(string,k))
    정렬출력위한_리스트 = []
    for i in 퍼뮤테이션_결과리스트:
        정렬출력위한_리스트.append("".join(i))
    정렬출력위한_리스트.sort()
    for i in 정렬출력위한_리스트:
        print(i)
if __name__ == "__main__":
    입력값  = input().split()
    퍼뮤테이션(입력값[0], int(입력값[1]))