Giganto cluster에서 Publish (stream 방식) 통신 지원

[sophie-cluml]

Background

#327 (comment)

이 이슈에서 임시로 사용할 약어

• "마스터 giganto" == hog/crusher의 config 파일에 적힌 giganto_addr 위치에 존재하는 giganto
• "pig124578" == vec!["pig1","pig2","pig4","pig5","pig7","pig8"]

고려할 상황들 - CORE

• Case 1 : Giganto 1-3 에게 Piglet 1-9들이 연결되어 있는 상태에서 Hog가 Giganto에게 source = "all" 에 대한 스트리밍 요청을 하는 경우 ("all" 은 모든 source 전체를 스트리밍 해달라고 요청하는 우리 코드 상의 magic string)
• Case2 : Giganto 1-3 에게 Piglet 1-9들이 연결되어 있는 상태에서 Hog가 Giganto에게 source = "pig124578" 에 대한 스트리밍 요청을 하는 경우
• Case3 : T0 시점에는 Giganto가 2개, Piglet이 4개 운영되고 있는 상황에서 source ="all" 스트리밍 요청을 대응하고 있다가, T1 시점에 Giganto 1대 추가, Piglet 2개 추가되는 경우

고려할 상황들 - EDGE

• Case1 : Giganto에게 Piglet들이 연결되지 않은 상태에서 Hog가 Giganto에게 source = "all" 에 대한 스트리밍 요청을 하는 경우
  • 기대되는 동작 : 추후 piglet 이 giganto에 연결되면 그때부터 바로 스트리밍 전송이 시작됩니다.
• Case2 : Giganto에게 Piglet들이 연결되지 않은 상태에서 Hog가 Giganto에게 source = "pig124578" 에 대한 스트리밍 요청을 하는 경우
  • 기대되는 동작 : 추후 piglet 이 giganto에 연결되면 그때부터 바로 source에 해당되는 것들에 한해 스트리밍 전송이 시작됩니다.
• Case3 : T0시점에는 Gignato 1번이 Piglet 3번과 연결되어있었고, T1 시점 이후로는 쭉 Giganto 3번이 Piglet 3번과 연결되는 상황이고, T2 시점에 Crusher가 source="pig3" 에 대해 start_time = T0 에 대한 스트리밍 요청을 하는 경우
  • 기대되는 동작 : Giganto 1번의 자료와 Giganto 3번의 자료가 스트리밍 됩니다.
  • 기대되는 동작(필수인지 확인 필요) : Giganto 1번의 자료가 먼저 스트리밍 되고, Giganto 3번의 자료가 스트리밍 됩니다.
• Case4 : Giganto 1-3 에게 Piglet 1-9들이 연결되어있는 조합이 변경되는 상황. (ex. T0 시점에는 Gig1로 pig1-3, Gig2로 pig4-6, Gig3으로 pig7-9 로 연결되어 있다가, T1 시점에는 Gig1이 pig4-6, Gig2가 pig1-3으로 변경된 상황)

구현 방향 아이디어

<구현 방식 2가지>
현시점 2가지 방식 구현이 논의되었습니다. 장단점 및 현시점의 코드 상황에 기반한 논의이므로, 구현 시점에 재확인이 필요합니다.

<방식 1> Hog/Crusher가 가능한 모든 Giganto와의 Multiple connection을 형성하고 유지하는 방식

• "마스터 giganto"는 hog/crusher와 최초 연결 시, giganto peer들의 socket addr 정보를 최초 전달하고, 그 이후 peer 변경이 생길 때마다 업데이트를 전달해주는 stream-endpoint-info-connection (이하 "스트리밍정보커넥션") 을 형성

• hog/crusher는 "스트리밍정보커넥션"을 통해 전달받은 socket addr에 대해서 전부 동일한 stream_request를 전송.

• 방식1에 대한 작성자의 의견

  • (장)구현과 아이디어가 상대적으로 간단함
  • (장) CORE-Case1,2,3 및 EDGE-Case1,2,4가 커버됨 (한계: EDGE-Case4의 경우, 잠시동안 시간 순서가 지켜지지 않는 순간 발생 가능; 이를 막기 위해선 giganto-piglet 조합 변경 시 n초 정도의 간격을 두는 오퍼레이션 필요한데, piglet off -> on 하는 동안 충분히 n초 지날 것으로 보임)
  • (단) hog/crusher가 giganto 개수만큼 항상 연결하게 됨

<방식 2> Hog/Crusher가 가능한 최소한의 Giganto와의 Multiple connection을 형성하고 유지하는 방식

• "마스터 giganto"는 hog와 최초 연결 시, giganto-source(piglet) 매핑 정보를 최초 전달하고, 그 이후 peer 변경 또는 source 변경이 생길 때마다 업데이트를 전달해주는 source-mapping-info-connection (이하 "매핑커넥션") 을 형성

• hog/crusher의 stream 전송 요청 발생

  • CASE source = "all" 요청 : "마스터 giganto"로 들어오는 모든 "all"에 대해 스트리밍 전송 && 모든 giganto peer들의 socket addr 정보 및 "all" 을 전달 && 이를 전달받은 hog/crusher는 전달받은 socket addr에 대해서 "all" 요청 전송
  • CASE source = "pig124578" 요청 : "마스터 giganto"가 담당하는 source인 "pig1", "pig2"에 대해서는 직접 스트리밍 전송 && 해당하는 giganto peer 들의 socket addr 정보 및 peer 별 source를 매핑하여 전달 && 이를 전달받은 hog/crusher는 전달받은 socket addr에 대해서 각 source 매핑 정보를 기반으로 새로운 요청을 만들어서 전송
  • CASE source = ["pig1"] 요청 : "마스터 giganto"가 담당하는 source 이므로, 직접 스트리밍 전송

• 방식2에 대한 작성자의 의견

  • (장) hog/crusher가 필요한 만큼 연결됨
  • (장) CORE-Case1,2,3 및 EDGE-Case1,2,4가 커버됨 (한계: EDGE-Case4의 경우, Giganto 변경 이후의 자료의 앞부분 일부를 못 받는 상황 발생 가능; 이를 막기 위해선 추가적인 코드 변경 필요)
  • (단) 구현과 아이디어가 상대적으로 복잡함

EDGE-Case3 에 대한 고려

EDGE-Case3 을 커버하기 위해선 방식 1, 2 어떤 것을 택하든 과거 giganto와 연결되었던 source 히스토리를 확인할 수 있는 기능이 필요합니다. (구현방식이 인메모리 히스토리 테이블이든, db key 검색이든, 어떤 방법이든 간에)

[kimhanbeom]

현재 crusher의 stream 요청에 대한 처리 수신부는 항상 시간순으로 들어온다는 가정하에 처리해서 시계열을 생성하고 있습니다.
위의 2가지 방식으로 crusher가 복수의 giganto로 stream을 요청하는 경우, 약간의 시간 순서 차이가 있을 수 있는 실시간 데이터의 경우에는 시계열 생성에 문제가 없겠지만, rocksdb의 저장된 데이터의 경우에는 각 giganto 별 데이터의 시간 차이가 클 가능성도 있으므로 올바른 시계열 생성이 이뤄질 수 없을 것으로 보입니다.

위의 상황을 고려해서 crusher의 cluster publish를 지원하기 위한 구현 아이디어를 생각해 봤는데 의견 부탁 드립니다.

• giganto 에서 stream 요청 수신 시 저장 데이터 전송하고 더 이상 저장 데이터가 없는 경우에는 끝을 나타내는 값(timestaamp,i64::MAX)을 전송, 그 후에 실시간 데이터를 전송합니다.
• crusher 는 복수의 giganto로부터 저장데이터를 수신해서 저장하고, 모든 giganto로부터 저장 데이터의 전송 끝(timestaamp,i64::MAX)을 받은 경우, 인메모리에 저장된 값으로 time series 생성, 그 후의 수신되는 값은 실시간 데이터로 판단하고 기존의 로직대로 처리 합니다.

[sophie-cluml]

좋은 의견인 것 같아요. 인메모리로 들고 있을 수 있는 규모를 넘어설 경우를 대비해서 살짝 변형할 수도 있을것 같습니다.

• 상황의 가정

  • source="pig1" 에 대하여 Gig1에는 "pig1"의 timestamp 001 ~ 099 데이터가 db에 있는 상태, Gig2에는 "pig1"의 timestamp 200 ~ 499 데이터가 있는 상태, Gig3에는 timestamp 100 ~ 199 데이터도 있고, 현시점에서 "pig1" 데이터가 500 ~ ongoing 실시간으로 들어오고 있는 상태
  • 그리고 crusher의 timestamp 001 ~ 499 에 해당하는 데이터를 메모리에 다 들고 있는 하드웨어 성능이 안 되는 상태

• 변형된 방법

  • crusher가 source ="pig1" 에 대한 request stream 을 "마스터 giganto"에게 전송
  • "마스터 giganto"는 모든 giganto에게 "pig1"에 데이터의 db 상 존재여부 및 현재 connection 상태에 대한 메타정보 질의 전송
  • Gig1은 timestamp (001, 099) 메타정보를 전송, Gig2 는 (200, 500) 메타정보를 전송, Gig3는 (100, ongoing) 메타정보를 전송
  • "마스터 giganto"는 메타정보 3가지를 조합하여 db 상 이미 저장된 값으로 판단되는 (001 ~ 500) 구간에 대하여
    • crusher가 순차적으로 (001 ~ 099) 는 Gig1에게 요청하고, (100 ~ 199) 는 Gig3에게, (200 ~ 500)은 Gig2에게 요청해야 한다는 지시를 담은 struct 를 crusher에게 전송
  • crusher는 지시에 따라서 순차적으로 데이터를 stream 수신받음
  • 그 후 crusher는 timestamp 500 이후의 데이터만 달라고 하는 요청을 giganto에게 전송하고, giganto는 기존처럼 처리