[프로젝트 수행] k8s 환경에서 프로젝트 ELK 적용 & 연동

로컬환경에서 기능 테스트

커스텀한 플젝을 로컬에서 테스팅

• server : Python-Flask를 사용한 시그널링 서버
• Front : React로 클라이언트를 분리
  새로고침을 해야만 화면공유와 채팅이 되어서 동시성 문제를 해결하기 위해 react로 분리하였다

• 채팅 및 화상통화 성공

ELK 적용전 서버 코드

데모 테스트로 작성한 프로젝트는 ELK연동을 성공했었다
이제는 메인 플젝에 ELK 적용을 시켜야 한다

ELK 적용전 코드

• app.py

from flask import Flask, render_template, request, session
from flask_socketio import SocketIO, emit, join_room, send
from flask_cors import CORS


app = Flask(__name__)
CORS(app)
app.config['SECRET_KEY'] = "test key"
socketio = SocketIO(app, cors_allowed_origins="*")

users_in_room = {}
rooms_sid = {}
names_sid = {}

@app.route('/')
def hello():
    return 'hello'

def messageReceived():
    print('message was received!!!')

# @socketio.on('connect') ################### test
# def test_connect():
#     print("goooooooooooooooooooooooooooooooood")
#     emit('pong')
    
@socketio.on('ping')
def test_ping():
    print('ponggggggggg')


@app.route("/join", methods=["GET"])
def join():
    display_name = request.args.get('display_name')
    mute_audio = request.args.get('mute_audio') # 1 or 0
    mute_video = request.args.get('mute_video') # 1 or 0
    room_id = request.args.get('room_id')
    session[room_id] = {"name": display_name,
                        "mute_audio": mute_audio, "mute_video": mute_video}
    return render_template("join.html", room_id=room_id, display_name=session[room_id]["name"], mute_audio=session[room_id]["mute_audio"], mute_video=session[room_id]["mute_video"])

@socketio.on("create-room")
def on_create_room(data):
    session[data["room_id"]] = {
        "name": data["display_name"],
        "mute_audio": data["mute_audio"],
        "mute_video": data["mute_video"]
    }
    print(session)
    emit("join-request")

@socketio.on("connect")
def on_connect():
    sid = request.sid
    print("New socket connected ", sid)


@socketio.on("join-room")
def on_join_room(data):
    sid = request.sid
    room_id = data["room_id"]
    display_name = session[room_id]["name"]
    print(sid)
    
    # register sid to the room
    join_room(room_id)
    rooms_sid[sid] = room_id
    names_sid[sid] = display_name
    
    # broadcast to others in the room
    print("[{}] New member joined: {}<{}>".format(room_id, display_name, sid))
    emit("user-connect", {"sid": sid, "name": display_name},
        broadcast=True, include_self=False, room=room_id)
    # broadcasting시 동일한 네임스페이스에 연결된 모든 클라이언트에게 메시지를 송신함
    # include_self=False 이므로 본인을 제외하고 broadcasting
    # room=room_id인 room에 메시지를 송신합니다. broadcast의 값이 True이어야 합니다.
    print("user-connect active")
    # add to user list maintained on server
    if room_id not in users_in_room:
        users_in_room[room_id] = [sid]
        emit("user-list", {"my_id": sid})  # send own id only
    else:
        usrlist = {u_id: names_sid[u_id]
                   for u_id in users_in_room[room_id]}
        # send list of existing users to the new member
        print(usrlist)
        emit("user-list", {"list": usrlist, "my_id": sid})
        # add new member to user list maintained on server
        users_in_room[room_id].append(sid)

    print("\n users: ", users_in_room, "\n")


@socketio.on("disconnect")
def on_disconnect():
    sid = request.sid
    room_id = rooms_sid[sid]
    display_name = names_sid[sid]

    print("[{}] Member left: {}<{}>".format(room_id, display_name, sid))
    emit("user-disconnect", {"sid": sid},
         broadcast=True, include_self=False, room=room_id)

    users_in_room[room_id].remove(sid)
    if len(users_in_room[room_id]) == 0:
        users_in_room.pop(room_id)

    rooms_sid.pop(sid)
    names_sid.pop(sid)

    print("\nusers: ", users_in_room, "\n")


@socketio.on("data")
def on_data(data):
    sender_sid = data['sender_id']
    target_sid = data['target_id']
    if sender_sid != request.sid:
        print("[Not supposed to happen!] request.sid and sender_id don't match!!!")

    if data["type"] != "new-ice-candidate":
        print('{} message from {} to {}'.format(
            data["type"], sender_sid, target_sid))
    socketio.emit('data', data, room=target_sid)

@socketio.on("chatting")
def message(message):
    #sid = request.sid
    #room_id = message["room_id"]
    #display_name = session[room_id]["name"]

    # register sid to the room
    #rooms_sid[sid] = room_id
    #names_sid[sid] = display_name
    print(message)
    
    # broadcast to others in the room
    emit("chatting", message , broadcast=True, include_self=False)

if any(platform.win32_ver()):
    socketio.run(app, debug=True)

ELK 적용

---

• elasticsearck docker 이미지 pull

docker pull docker.elastic.co/elasticsearch/elasticsearch:7.10.1

• 컨테이너 실행

docker run -d -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:7.10.1

• kibana docker 이미지 pull

docker pull docker.elastic.co/kibana/kibana:7.10.1

• 컨테이너 실행

docker run -d --link [elasticsearch 컨테이너 id]:elasticsearch -p 5601:5601 docker.elastic.co/kibana/kibana:7.10.1

ELK 적용 코드

app.py

from flask import Flask, render_template, request, session
from flask_socketio import SocketIO, emit, join_room, send
from elasticsearch import Elasticsearch
from elasticsearch import helpers
import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = "test key"
socketio = SocketIO(app, cors_allowed_origins="*")

users_in_room = {}
rooms_sid = {}
names_sid = {}

### elk, kibana
es = Elasticsearch('http://192.168.56.141:9200') ## 변경
es.info()

def utc_time():  
    return datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'

def make_index(es, index_name):
    if es.indices.exists(index=index_name):
        es.indices.delete(index=index_name)
        es.indices.create(index=index_name)

index_name= 'webrtc_room'

@app.route('/')
def hello():
    return 'hello'

@socketio.on('connect') ################### test
def test_connect():
    print("connection is successs")

@app.route("/join", methods=["GET"])
def join():
    return render_template("join.html")

@socketio.on("create-room")
def on_create_room(data):
    session[data["room_id"]] = {
        "name": data["display_name"],
        "mute_audio": data["mute_audio"],
        "mute_video": data["mute_video"]
    }
    print(session)
    emit("join-request")

 # elk
    room_id = data["room_id"]
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_create= {"des":"create room", "room_id":room_id, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_create)

@socketio.on("join-room")
def on_join_room(data):
    sid = request.sid
    room_id = data["room_id"]
    display_name = session[room_id]["name"]
    
    # register sid to the room
    join_room(room_id)
    rooms_sid[sid] = room_id
    names_sid[sid] = display_name
    # broadcast to others in the room
    print("[{}] New member joined: {}<{}>".format(room_id, display_name, sid))

### elk
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_join= {"des":"New member joined", "room_id":room_id, "sid": sid, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_join)   
   
   
    emit("user-connect", {"sid": sid, "name": display_name},
        broadcast=True, include_self=False, room=room_id)
    # broadcasting시 동일한 네임스페이스에 연결된 모든 클라이언트에게 메시지를 송신함
    # include_self=False 이므로 본인을 제외하고 broadcasting
    # room=room_id인 room에 메시지를 송신합니다. broadcast의 값이 True이어야 합니다.
    # add to user list maintained on server
    if room_id not in users_in_room:
        users_in_room[room_id] = [sid]
        emit("user-list", {"my_id": sid})  # send own id only
    else:
        usrlist = {u_id: names_sid[u_id]
                   for u_id in users_in_room[room_id]}
        # send list of existing users to the new member
        print(usrlist)
        emit("user-list", {"list": usrlist, "my_id": sid})
        # add new member to user list maintained on server
        users_in_room[room_id].append(sid)

    print("\n users: ", users_in_room, "\n")

# leave_room은 사용하지 않아도 되는지?

@socketio.on("disconnect")
def on_disconnect():
    sid = request.sid
    room_id = rooms_sid[sid]
    display_name = names_sid[sid]

### elk
    now = datetime.datetime.now()
    now = now.strftime('%m/%d/%y %H:%M:%S')
    doc_disconnect= {"des":"user-disconnect", "room_id":room_id, "sid": sid, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_disconnect)

    print("[{}] Member left: {}<{}>".format(room_id, display_name, sid))
    emit("user-disconnect", {"sid": sid},
         broadcast=True, include_self=False, room=room_id)

    users_in_room[room_id].remove(sid)
    if len(users_in_room[room_id]) == 0:
        users_in_room.pop(room_id)

    rooms_sid.pop(sid)
    names_sid.pop(sid)

    print("\nusers: ", users_in_room, "\n")

@socketio.on("data")
def on_data(data):
    sender_sid = data['sender_id']
    target_sid = data['target_id']
    if sender_sid != request.sid:
        print("[Not supposed to happen!] request.sid and sender_id don't match!!!")

    if data["type"] != "new-ice-candidate":
        print('{} message from {} to {}'.format(
            data["type"], sender_sid, target_sid))
    socketio.emit('data', data, room=target_sid)

@socketio.on("chatting")
def send_message(message):
    sender = message["sender"]
    text = message["text"]
    room_id = message["room_id"]

### elk
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_chatting= {"des" : "chatting", "room_id" : room_id, "sid" : sid, "chatting message" : message,"@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_chatting)

    # broadcast to others in the room
    emit("chatting", message , broadcast=True, include_self=True, room=room_id)

if __name__ == '__main__':
    socketio.run(app,
        host="0.0.0.0",
        port=5000,
        debug=True 
        #ssl_context=("cert.pem", "key.pem")
    )
    make_index(es, index_name)

코드 설명

• 관련 라이브러리 호출

from elasticsearch import Elasticsearch
from elasticsearch import helpers
import datetime

• elasticsearcch 연결, index 생성 함수 정의, index 이름 생성

### elk, kibana
es = Elasticsearch('http://192.168.56.141:9200') ## 변경
es.info()

def utc_time():  
    return datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'

def make_index(es, index_name):
    if es.indices.exists(index=index_name):
        es.indices.delete(index=index_name)
        es.indices.create(index=index_name)

index_name= 'webrtc_room'

create room의 로그 생성, 로그 삽입

# elk
    room_id = data["room_id"]
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_create= {"des":"create room", "room_id":room_id, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_create)

• new member joined의 로그 생성, 로그 삽입

### elk
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_join= {"des":"New member joined", "room_id":room_id, "sid": sid, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_join)   

• user-disconnect의 로그 생성, 로그 삽입

### elk
    now = datetime.datetime.now()
    now = now.strftime('%m/%d/%y %H:%M:%S')
    doc_disconnect= {"des":"user-disconnect", "room_id":room_id, "sid": sid, "@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_disconnect)

• chatting의 로그 생성, 로그 삽입

### elk
    date = datetime.datetime.now()
    now = date.strftime('%m/%d/%y %H:%M:%S')
    doc_chatting= {"des" : "chatting", "room_id" : room_id, "sid" : sid, "chatting message" : message,"@timestamp": utc_time()}
    es.index(index=index_name, doc_type="log", body=doc_chatting)

• index 생성

if __name__ == '__main__':
    socketio.run(app,
        host="0.0.0.0",
        port=5000,
        debug=True 
        #ssl_context=("cert.pem", "key.pem")
    )
    make_index(es, index_name) <-- 요거

ELK 접속

• ELK 접속: {serverIP}:5601
  
• index pattern 생성
  위의 코드가 잘 적용되었다면 index가 생성되어 있어 Next Step 버튼을 누를 수 있다
  
• Timestamp 적용
  시간별 정보를 수집해야하기 때문에 해당 부분을 같이 선택해준다
  
• index 생성후
  

visualization

시간, room별 서비스 사용량 데이터 만들기

하루동안 서비스 사용량 데이터 만들기
metric에 들어가서 하기