클라우드 개발자 앙몬드

오랜만에 파이썬 다시 꺼냈음.

venv좀 써야할 일이 생김..... pythonflow

환경은 맥 기준

venv 만들기

# python3 -m venv ./가상환경명 

python3 -m venv ./vvenv

venv 활성화

# source 가상환경명/bin/activate 

source vvenv/bin/activate

venv 비활성화

deactivate

리얼타임. 실시간 성으로 데이터를 처리 시, 공간복잡도를 크게 고려하지 않아도 되었으나, 현재 초 대용량 (목표치는 약 5.2 천만건 Rows) 데이터를 파일로 내리고 파싱할 일이 생겼습니다.

기본적으로 파이썬에서는 이터레이터로 생성하고, 실제 그 값이 필요해지는 시점에서, 그 값을 만드는 방식을 사용하는데요.

이전 C언어의 포인터, 그리고 자바의 레퍼런스를 빌어 사용하자면, 반복해야하는 콜렉션에 대한 메모리 주소 및 리모콘만 만들어두고, 그 내용물에 해당하는 데이터 생성을 뒤로 미루는 것이죠. 이런 기법을 지연 평가(lazy evaluation)라고 전문적으로 부른다고 하네요 ㅇㅇ..

유사 사례로, 브라우저에서 웹 이미지, CSS 등 스태틱 컨텐츠들을 가져올 시, 레이지 로딩 (Lazy Load) 라는 기법 또한 존재합니다. 먼저, 이 스태틱 URI들을 브라우저가 받은 뒤, 그 스태틱 이미지들을 이후에 CDN에서 가져오는 것이죠 :) 

(나머지 공사중)

S3 전체가 약 50페타.

그런 버킷이 8세트...

전체를 크롤링 시, 프리픽스가 없는 상태라면 어쩔 수 없이 시간싸움으로 가게 된댜....

import boto3, json, pathlib, sys, asyncio
from datetime import datetime

class S3Util :
	def __init__(self, 분석할_버킷명, 파일명, 게임id=None, 테이블명=None, 연=None, 월=None, 일=None) :
		self.버킷명 = 분석할_버킷명
		self.게임id, self.테이블명 = 게임id, 테이블명 
		self.연, self.월, self.일 = 연, 월, 일
		self.파일명 = 파일명
	
	# s3 오브젝트 크롤링
	def s3_오브젝트_크롤링(self, prefix=None) :
		pathlib.Path("./result").mkdir(parents=True, exist_ok=True)

		s3c_lister = boto3.client("s3").get_paginator("list_objects_v2")
		페이저 = s3c_lister.paginate(Bucket = self.버킷명)
		if prefix != None : 페이저 = s3c_lister.paginate(Bucket = self.버킷명, Prefix=prefix)
		
		with open("./result/" + self.파일명, "w") as 로컬_원본파일 : 
			for 페이지 in 페이저 :
				for 응답 in 페이지["Contents"] :
					응답키 = 응답["Key"].replace("/", ",")
					로컬_원본파일.write( ",".join( ( 응답키, str(응답["Size"]), "\n" ) ) )
	
	# s3 오브젝트 크롤링 비동기 업그레이드 
	def s3_오브젝트_비동기_크롤링(self, prefix=None) :
		pathlib.Path("./result").mkdir(parents=True, exist_ok=True)

		s3c_lister = boto3.client("s3").get_paginator("list_objects_v2")
		페이저 = s3c_lister.paginate(Bucket = self.버킷명)
		if prefix != None : 페이저 = s3c_lister.paginate(Bucket = self.버킷명, Prefix=prefix)

		async def _s3_obj_list_rslt_file_write(파일, 응답키, 응답) :
			파일.write(",".join( ( 응답키, str(응답["Size"]), "\n" ) ))

		async def _s3_obj_list_async(파일) :
			async for 페이지 in 페이저 :
				for 응답 in 페이지["Contents"] :
					응답키 = 응답["Key"].replace("/", ",")
					await _s3_obj_list_rslt_file_write(파일, 응답키, 응답)

		with open("./result/" + self.파일명, "w") as 로컬_원본파일 :
			이벤트루프 = asyncio.get_event_loop()
			이벤트루프.run_until_complete(_s3_obj_list_async(로컬_원본파일))

	# 날짜별 총합
	def s3_크롤링오브젝트_날짜별총합(self, 취합할_파일명) :
		취합결과 = { "dummy" : "dummy" }
		with open('./result/' + 취합할_파일명, 'r') as 파일:
			for 파일한줄 in 파일 :
				스플릿된_라인 = 파일한줄.split(",")
				try : 
					yyyyMMdd = "-".join((스플릿된_라인[2], 스플릿된_라인[3], 스플릿된_라인[4]))
					이_파일의_용량 = int(스플릿된_라인[-2])
					try : 취합결과[yyyyMMdd] += 이_파일의_용량 
					except Exception as e : 취합결과[yyyyMMdd] = 이_파일의_용량
				except Exception as e :
					print("날짜별 파싱 오류 ")
					print(파일한줄)
	
		날짜출력_리스트 = []
		with open("yyyyMMddSum_" + 취합할_파일명 +".txt", "w") as file :
			for key in 취합결과.keys():
				날짜출력_리스트.append(key)

			날짜출력_리스트.sort()
			for element in 날짜출력_리스트 :
				쓸_내용 = ",".join( [ element, str( 취합결과[element] ) ] )
				file.write( 쓸_내용 + "\n" )

	# 게임id별 총합
	def s3_크롤링오브젝트_게임id별총합(self, 취합할_파일명) :
		취합결과 = { "dummy" : "dummy" }

		with open('./result/' + 취합할_파일명, 'r') as 파일:
			for 파일한줄 in 파일 :
				스플릿된_라인 = 파일한줄.split(",")
				게임id = str(스플릿된_라인[0])
				이_파일의_용량 = int(스플릿된_라인[-2])
				try : 취합결과[게임id] += 이_파일의_용량 
				except Exception as e : 취합결과[게임id] = 이_파일의_용량

		with open("gameIdSum_" + 취합할_파일명 +".txt", "w") as file :
			for k, v in 취합결과.items():
				쓸_내용 = ",".join( [ k, str(v) ] )
				file.write( 쓸_내용 + "\n")

	# 게임id-테이블id별 총합
	def s3_크롤링오브젝트_게임id_테이블별총합(self, 취합할_파일명) :
		취합결과 = { 
			"dummy" : { 
				"a" : 0 
			}}

		with open('./result/' + 취합할_파일명, 'r') as 파일:
			for 파일한줄 in 파일 :
				스플릿된_라인 = 파일한줄.split(",")
				게임id = str(스플릿된_라인[0])
				테이블id = str(스플릿된_라인[1])
				이_파일의_용량 = int(스플릿된_라인[-2])

				try : 취합결과[게임id] 
				except Exception as e : 취합결과[게임id] = {}

				try : 취합결과[게임id][테이블id] += 이_파일의_용량
				except Exception as e : 취합결과[게임id][테이블id] = 이_파일의_용량

		with open("gameId_and_tableId_Sum_" + 취합할_파일명 +".txt", "w") as file :
			for k, v in 취합결과.items():
				for x, y in v.items() :
					쓸_내용 = ",".join( [ k, x, str(y) ] )
					file.write( 쓸_내용 + "\n")
		
	# 게임id-테이블id-날짜별 총합
	def s3_크롤링오브젝트_게임id_테이블id_날짜별총합(self, 취합할_파일명) :
		취합결과 = {
			"gameid" : {
				"tableid" : {
					"2021-01-01" : 0
				}}}

		with open('./result/' + 취합할_파일명, 'r') as 파일:
			for 파일한줄 in 파일 :
				스플릿된_라인 = 파일한줄.split(",")
				게임id = str(스플릿된_라인[0])
				테이블id = str(스플릿된_라인[1])
				try : 
					yyyyMMdd = "-".join((스플릿된_라인[2], 스플릿된_라인[3], 스플릿된_라인[4]))
					이_파일의_용량 = int(스플릿된_라인[-2])

					try : 취합결과[게임id] 
					except Exception as e : 취합결과[게임id] = {}

					try : 취합결과[게임id][테이블id]
					except Exception as e : 취합결과[게임id][테이블id] = {}

					try : 취합결과[게임id][테이블id][yyyyMMdd] += 이_파일의_용량
					except Exception as e : 취합결과[게임id][테이블id][yyyyMMdd] = 이_파일의_용량 
				except Exception as e : 
					print("날짜 파싱 오류 데이터 발생")
					print(파일한줄)

		with open("gameId_and_tableId_and_yyyyMMdd_Sum_" + 취합할_파일명 +".txt", "w") as file :
			for k, v in 취합결과.items():
				for x, y in v.items() :
					for z, w in y.items() :
						쓸_내용 = ",".join( [k, x, z, str(w)] )
						file.write(쓸_내용 + "\n")

if __name__ == "__main__" : 
	버킷명, 게임id, 테이블id = None, None, None

	try :
		버킷명 = sys.argv[1]
		게임id = sys.argv[2]
		테이블id = sys.argv[3]
	except Exception as e :
		if 버킷명 == None :
			print("버킷명 필수입니다.")
			exit(-1)
		if 게임id == None and 테이블id != None :
			print("게임id 없이 테이블id로 크롤링할 수 없습니다.")
			exit(-1)
	
	# (1) 파일로 결과물을 출력 시, 이에 해당하는 날짜로 네이밍을 완성합니다.
	#		또한, 결과물을 해당 경로의 result에 디렉토리로 만듭니다.
	#		실행 시간 측정용으로, exeinfo 파일에 실행/종료 시간을 씁니다.
	today = str(datetime.today()).replace(" ", "_")
	파일명 = 버킷명 + "__" + str(게임id) + "__" + str(테이블id) + "__" + today

	pathlib.Path("./result").mkdir(parents=True, exist_ok=True)
	로컬_실행결과파일 = open("./result/" + 버킷명 + "_exeinfo_" + today, "w")
	로컬_실행결과파일.write( f"실행 시작시간 : '{today}' \n" )

	# (2) 버킷명 / 게임id / 테이블id 아규먼트를 기준으로, s3를 크롤링합니다.
	데이터분석 = S3Util(버킷명, 파일명, 게임id, 테이블id)
	if 게임id == None : 데이터분석.s3_오브젝트_크롤링() 
	elif 게임id != None and 테이블id == None : 데이터분석.s3_오브젝트_크롤링(게임id)
	elif 게임id != None and 테이블id != None : 데이터분석.s3_오브젝트_크롤링(게임id + "/" + 테이블id)
	else : print("입력값 분기처리 오류")

	# (3) 종료 시간을 작성 후, 파일스트림 파이프를 닫습니다.
	로컬_실행결과파일.write( f"실행 종료시간 : '{today}' \n" )
	로컬_실행결과파일.close()

노드js에서나 보던 비동기 처리를, 파이썬에서도 보게 될 줄이야...

파일쓰기에서 비동기 처리가 필요하였기에 테스트성으로 개발한 용도입니다.

import boto3, json, pathlib, sys, asyncio, logging, itertools
from datetime import datetime
from aiofile import AIOFile, Reader, Writer, async_open

async def 내용을_생성하고_파일쓰기_호출하는_함수() :
	for i in range(10000) :
		한페이지 = []
		for j in range(10000) : 한페이지.append(str(i) + "\n")
		asyncio.create_task(파일쓰기함수(한페이지))
		#await asyncio.create_task(파일쓰기함수(한페이지))
	 
async def 파일쓰기함수(한페이지) :
	with open("temp2.txt", "a") as 파일 : 파일.writelines(한페이지)

async def 주황버섯의_함수() :
	print("시작!")
	시작시간 = datetime.now()
	await asyncio.create_task(내용을_생성하고_파일쓰기_호출하는_함수())
	끝난시간 = datetime.now()
	print(f"{시작시간} -> {끝난시간}")
	print(f"총 걸린시간 : {끝난시간 - 시작시간}")

def 핑크빈의_파일쓰기_검증함수() :
	print("검증몬 출동")
	검증결과 = { }
	with open("temp2.txt", "r") as 파일 :
		한줄 = 파일.readline()
		while 한줄 != "" :
			try : 
				검증결과[한줄] += 1
			except Exception as e :
				검증결과[한줄] = 1
			한줄 = 파일.readline()

	for k,v in 검증결과.items() : 
		if v != 10000 : print(f"{k} : {v}")
		#print(f"{k} : {v}")

if __name__ == "__main__" :

	#핑크빈의_파일쓰기_검증함수()
	#exit(0)

	물범의_이벤트루프 = asyncio.get_event_loop()
	물범의_이벤트루프.run_until_complete(주황버섯의_함수())
	물범의_이벤트루프.close()

import pandas as pd

d = [{'모듈': '주문API', '상태': 'Terminating'}, {'모듈': '전시프론트', '상태': 'Terminating'},{'모듈': '전시프론트', '상태': 'Terminating'}, {'모듈': '주문API', '상태': 'Terminating'}]

d_unique = pd.DataFrame(d).drop_duplicates().to_dict('records')

print(d_unique)
# pip3 install pandas  !!!!!

pandas 라이브러리를 이용하여, 리스트 엘리먼트가 딕셔너리 타입일 때, 중복을 제거하는 방법입니다.

만약 람다 레이어 및 디펜던시 의존성 계층이 무거워서, 라이브러리 추가가 힘들 경우, 하기와 같은 심플한 방법으로 처리하는 것도 좋은 방법!!

list_of_data = [{'모듈': '주문API', '상태': 'Terminating'}, {'모듈': '전시프론트', '상태': 'Terminating'},{'모듈': '전시프론트', '상태': 'Terminating'}, {'모듈': '주문API', '상태': 'Terminating'}]
list_of_data_uniq = []
for data in list_of_data:
    if data not in list_of_data_uniq:
        list_of_data_uniq.append(data)

쿠버네티스 노드에 대한 이슈가 자꾸 불거지기에, 계속 요즘따라 관제를 커스터마이징 하고 있습니댜

이전에 터졌던 EKS PLEG이슈에 이어 간혹 산발적으로 노드가 하나씩 다운되는 사태까지..

머리속으로는 이해하지만, 막상 시간이 부족하여 잘 만들지 못했던 관제를 이제야 꺼내보는군여.....

사실 고대의 존재들이 만들어두었던 방식을 제가 그대로 싹 흡수해서 만들게 된 케이스이기도 합니댜

아주 먼 옛날, 이때 만들면서 시간대역 때문에 시간 허비했던 그 관제시스템의 템플릿 버전이기도 하죠.

한편으론 감사합니다. 머나먼 고대의 존재들에게 이어받은 모든 지식을 토대로, 이렇게 제가 활용하고 있따는 것을 보면 훗훗..

(앙몬드가 가을타다보니 별 생각이 다 든댜)

devloper-angmond.tistory.com/6

상기 환경과는 다르게 지금의 환경은, 클라우드 리소스 접근이 어렵기 때문에, 젠킨스의 크론 빌드를 이용하여, 관제를 걸기로 하였습니댜

따라서, 준비되는 리소스는 다음과 같습니댜

0. 환경작업 패스

1. EC2 머신 (젠킨스 설치 / 크론탭으로 빌드 유발)

2. 크론탭 빌드 유발 시, 준비 되어야하는 파이썬3 스크립트

3. ES & 키바나 비주얼라이즈 with S3 Serverless Host

4. 노드가 비정상으로 감지되었을 시, 텔레그램으로 관제 발송

간단한 아키텍처는 이와 같습니댜

텔레그램에 대한 관제는 정말 별거 없지만, 가이드가 없으면 갈피잡기가 너무 어렵기 떄문에, 훗날 서브로 다시 작성해둘게요.

"""
텔레그램 메세지 전송 헬퍼

@author 클라우드 개발자 앙몬드
@since 2020-11-02

[2020-11-02] 최초생성
"""

import telegram

토큰 = '봇에게 접근하기 위한 토큰값입니다. 숫자:문자 로 이루어져있어요'
#텔레그램_채팅방_아이디 = "채팅방마다 시퀀스가 존재합니다"  #내 개인방
텔레그램_채팅방_아이디 = "채팅방마다 시퀀스가 존재합니다"   #데브옵스 엔지니어 방 
봇 = telegram.Bot(token=토큰)

#채팅방 아이디를 얻어내기 위한, json값 획득
#for i in 봇.getUpdates():
#    print(i.message)

def 텔레그램_메세지보내기(메세지) :
    봇.sendMessage(chat_id=텔레그램_채팅방_아이디, text=메세지) 

# -*- coding: utf-8 -*- 
import json
import time
import os
import decimal
import sys
import subprocess
import yaml
import boto3 
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection
from datetime import datetime, timedelta
from os import rename, listdir
from angmond_es_helper import *
import telegram_helper

"""
쿠버네티스 클러스터 노드 상태 체크의 백엔드 로직

@since 2020-10-29
@author 클라우드 개발자 앙몬드

[2020-10-29] 최초생성
"""

비정상노드리스트 = []
ES입력리스트 = []

메세지내용 = ""
노드명 = ""
노드상태 = ""
노드_실행기간 = ""
노드그룹 = ""
today = datetime.today()

ES_노드상태_인덱스 = "devops-eks-node-status-"
ES_노드사용_인덱스 = "devops-eks-node-usage-"
ES_노드그룹사용_인덱스 = "devops-eks-nodegroup-usage-"

ES커넥터 = get_aws_es_connector(get_aws_es_credentials(True), "ES_호스트주소!")
ES타임스탬프 = (today - timedelta(hours=9)).isoformat()

if __name__ == "__main__" : 
    비정제_전체_노드_요약리스트 = subprocess.check_output("kubectl get node -L group | grep -v NAME", shell=True, universal_newlines=True).strip()
    노드리스트 = str(비정제_전체_노드_요약리스트).split("\n")

    노드_FO_CPU = []	#프론트 모듈중 CPU에 특화된 노드들이 구성된 노드그룹
    노드_FO_MEM = []	#프론트 모듈 중 메모리에 특화된 노드들이 구성된 노드그룹
    노드_BO_MEM = []	#백오피스/현업/ERP 모듈 중 메모리에 특화된 노드들이 구성된 노드그룹

    # (0) 비정상노드 탐지를 위한 정제부분
    완전정제된_노드리스트 = list(map(lambda x : (" ".join(x.split())).split(" "), 노드리스트))

    노드_FO_CPU = list(filter(lambda x : x[5] == "fo_cpu", 완전정제된_노드리스트))
    노드_FO_MEM = list(filter(lambda x : x[5] == "fo_mem", 완전정제된_노드리스트))
    노드_BO_MEM = list(filter(lambda x : x[5] == "bo_mem", 완전정제된_노드리스트))

    # (1) 비정상노드 상태값 판별 & 텔레그램 관제
    for 노드 in 완전정제된_노드리스트 :
        """
        노드정보는 리스트로..
        0 : NAME
        1 : STATUS
        2 : Role
        3 : AGE
        4 : EKS버전
        5 : 노드 그룹
        """  
        
        메모리부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  MemoryPressure'", shell=True, universal_newlines=True).strip()
        볼륨부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  DiskPressure'", shell=True, universal_newlines=True).strip()
        PID부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  PIDPressure'", shell=True, universal_newlines=True).strip()
        쿠블릿_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  Ready'", shell=True, universal_newlines=True).strip()
        
        관제메세지 = ""
        if "True" in 메모리부하_상태 :
            관제메세지 = 관제메세지 + "MemoryPressure 상태 감지 "
            메모리부하_상태 = True
        else : 메모리부하_상태 = False

        if "True" in 볼륨부하_상태 :
            관제메세지 = 관제메세지 + "DiskPressure 상태 감지 " 
            볼륨부하_상태 = True
        else : 볼륨부하_상태 = False

        if "True" in PID부하_상태 :
            관제메세지 = 관제메세지 + "PIDPressure 상태 감지 " 
            PID부하_상태 = True
        else : PID부하_상태 = False

        if "False" in 쿠블릿_상태 :
            관제메세지 = 관제메세지 + "쿠블릿 비정상 감지 " 
            쿠블릿_상태 = False
        else : 쿠블릿_상태 = True

        # 상태 이상 발견 시, 텔레그램 관제 문자 발송
        if len(관제메세지) != 0 :
            telegram_helper.텔레그램_메세지보내기(노드[0] + " -> " + 관제메세지)
        
        # ES인덱싱
        es_다큐먼트 = {
            "node" : 노드[0],
            "MemoryPressure" : 메모리부하_상태,
            "DiskPressure" : 볼륨부하_상태,
            "PIDPressure" : PID부하_상태,
            "KubeletReady" : 쿠블릿_상태,
            "@timestamp" : ES타임스탬프,
        }
        print(es_다큐먼트)
        ES인덱스 = ES_노드상태_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, es_다큐먼트)       

    # (2) 그룹별 CPU/Mem 사용률 
    비정제_노드자원사용률리스트 = subprocess.check_output("kubectl top node | grep -v NAME", shell=True, universal_newlines=True).strip()
    노드자원사용률리스트 = str(비정제_노드자원사용률리스트).split("\n")

    FO_CPU_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "fo_cpu", 완전정제된_노드리스트))))
    FO_MEM_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "fo_mem", 완전정제된_노드리스트))))
    BO_MEM_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "bo_mem", 완전정제된_노드리스트))))
    FO_CPU_리소스필터링_리스트 = []
    FO_MEM_리소스필터링_리스트 = []
    BO_MEM_리소스필터링_리스트 = []
    for 노드리소스사용률 in 노드자원사용률리스트:
        정제된_노드리소스_사용정보 = (" ".join(노드리소스사용률.split())).split(" ")        
        """
        [0] 노드명
        [1] CPU 사용량
        [2] CPU 사용률
        [3] Mem 사용량
        [4] Mem 사용률
        """

        es_다큐먼트 = {
            "node" : 정제된_노드리소스_사용정보[0],
            "cpu_usage" : int(정제된_노드리소스_사용정보[1].replace("m", "")),
            "cpu_use_rate" : int(정제된_노드리소스_사용정보[2].replace("%", "")),
            "mem_usage" : int(정제된_노드리소스_사용정보[3].replace("Mi", "")),
            "mem_use_rate" : int(정제된_노드리소스_사용정보[4].replace("%", "")),
            "@timestamp" : ES타임스탬프,
        }
        ES인덱스 = ES_노드사용_인덱스 + str(today.year) + '.' + str(today.month)
        print(es_다큐먼트)
        create_es_index(ES커넥터, ES인덱스, es_다큐먼트)

        if 정제된_노드리소스_사용정보[0] in FO_CPU_노드만 :
            es_다큐먼트["node_group"] = "fo_cpu"
            FO_CPU_리소스필터링_리스트.append(es_다큐먼트)
        elif 정제된_노드리소스_사용정보[0] in FO_MEM_노드만 :
            es_다큐먼트["node_group"] = "fo_mem"
            FO_MEM_리소스필터링_리스트.append(es_다큐먼트) 
        elif 정제된_노드리소스_사용정보[0] in BO_MEM_노드만 :
            es_다큐먼트["node_group"] = "bo_mem"
            BO_MEM_리소스필터링_리스트.append(es_다큐먼트) 

    for FO_CPU_정보 in FO_CPU_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, FO_CPU_정보)
    for FO_MEM_정보 in FO_MEM_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, FO_MEM_정보)
    for BO_MEM_정보 in BO_MEM_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, BO_MEM_정보)