Couchbase Eventing Handling Errors and Retries w/ Examples

그리고 카우치베이스 이벤트 서비스 를 사용하면 데이터의 돌연변이(또는 변경)에 대해 즉시 조치를 취할 수 있습니다. 모든 작업 이벤트 는 자바스크립트로 작성된 작은 비즈니스 로직인 람다를 실행하여 수행됩니다.

일반적인 사용 사례로는 데이터 강화, 문서 보관, 외부 REST 서비스와의 통합 등이 있습니다. 자세한 내용 보기 여기.

다음 블로그에서는 이벤트 리스너 실행 중 오류를 처리하는 방법에 대해 간략하게 설명합니다. 재시도 메커니즘을 사용하면 이벤트 리스너가 실행 중에 실패하더라도 예상된 작업이 수행되도록 보장합니다.

샘플 애플리케이션

예를 들어, 고객 주문을 Couchbase 컬렉션에 저장하는 전자상거래 애플리케이션의 일부를 구현했습니다. 주문 상태가 다음과 같이 변경되는 즉시 유료를 클릭하여 고객에게 주문 확인을 보내려고 합니다.

이를 위해 Couchbase 이벤트 리스너를 외부의 이메일 서비스. 카우치베이스 이벤트 리스너는 주문 문서의 변경 사항을 포착하고 주문이 결제되었는지 확인한 다음 이메일 서비스를 호출하여 확인 메시지를 트리거합니다.

이메일 서비스는 REST 엔드포인트를 제공하는 독립형 마이크로서비스입니다. 이벤트 리스너에서 마이크로서비스를 호출하기 위해 Couchbase Eventing에 직접 내장된 cURL 지원을 사용합니다.

개념적 흐름: 카우치베이스에서 주문이 업데이트되면 이벤트 리스너가 이벤트를 트리거하고 수집합니다. 그러면 이벤트 리스너가 외부 이메일 서비스로 호출합니다.

오류를 반환하는 이메일 서비스

설명한 시나리오는 다음과 같은 경우 매우 잘 작동합니다. 이메일 서비스 가 작동합니다. 하지만 이메일 서비스 가 오류를 반환하나요? 이벤트 수신기에서 이메일 서비스로의 요청이 실패하므로 고객에게 확인 메시지가 전송되지 않습니다. 실패 시점에 Couchbase 문서 변경 이벤트가 이미 처리되었으므로 다른 변경 사항이 없는 한 동일한 문서에 대한 새 이벤트가 트리거되지 않습니다. 확인이 전송되도록 하려면 오류를 처리하고 재시도 메커니즘을 구현해야 합니다. 이렇게 하면 외부 서비스의 일시적인 문제를 해결함과 동시에 확인이 전송되도록 보장할 수 있습니다.

이에 접근하는 방법에는 여러 가지가 있습니다. 아래 예제에서는 이벤트 리스너 실행이 실패한 문서에 대한 참조를 저장하는 '다시 시도'라는 새 컬렉션을 만들기로 선택했습니다.

이벤트 리스너는 주문 문서의 변경 사항을 수신한 다음(#1 단계) 이메일 서비스를 호출합니다(#2 단계). 이메일 서비스 호출이 성공하면 이벤트 리스너는 주문 문서의 확인 메시지 상태를 업데이트합니다(#3 단계). 그러나 실패할 경우 재시도 문서가 생성되어 '재시도' 컬렉션에 저장됩니다(#3* 단계).

문서에 대한 참조를 유지하면 실패한 모든 업데이트를 식별할 수 있고 나중에 다시 실행할 수 있습니다. 이는 운영자가 수동으로 개입하거나 Couchbase 이벤트 타이머를 사용하여 자동으로 재시도할 수 있습니다.

지정된 문서 ID를 가진 문서를 재시도 컬렉션에 추가하여 재시도 프로세스를 시작합니다. 제공된 타이머 간격에 따라 반복 타이머가 생성됩니다.
타이머 실행 시 작은 시간 퀀터보다 오래된 모든 문서는 다시 시도 컬렉션이 업데이트됩니다. 다음과 같은 속성을 추가하면 fireRetry = true 를 재시도 문서에 추가하면 이벤트 리스너가 포착한 또 다른 업데이트 이벤트를 트리거하여 재시도 메커니즘을 실행합니다. 이렇게 하면 재시도 컬렉션의 모든 문서에 불을 켜는 재귀적 돌연변이가 병렬로 실행됩니다. 이제 재시도 함수는 사용 가능한 모든 워커 스레드를 병렬로 사용하여 실행됩니다.
문서 업데이트 이벤트는 각 재시도 문서에 대해 개별적으로 트리거됩니다.
인바운드 컬렉션에서 해당 주문 문서가 검색됩니다.
이제 이메일 서비스가 호출됩니다.
이메일 서비스에 대한 호출이 성공하면 이벤트 수신기가 주문 문서의 확인 메시지 상태를 업데이트하고 재시도 문서를 제거합니다.
실패할 경우 재시도 문서가 업데이트되어 '재시도' 컬렉션에 저장됩니다.

코드 검토

이제 개념 설계를 설정했으니 샘플 구현을 살펴보겠습니다:

전제 조건:

카우치베이스 7 엔터프라이즈 에디션. 저는 로컬 컴퓨터의 Docker에서 단일 노드 클러스터로 Couchbase를 실행합니다. (https://docs.couchbase.com/server/current/install/getting-started-docker.html)
개발 목적으로 다음 서비스를 실행하는 단일 노드 카우치베이스 클러스터를 생성합니다:
- 인덱스, 쿼리, 이벤트 및 데이터 서비스

프로덕션 사용에는 단일 노드 설치를 권장하지 않습니다.

준비

라는 이름의 버킷을 만듭니다. 주문
에서 두 개의 컬렉션을 만듭니다. 주문 버킷 _기본 범위:
- 인바운드 (여기에는 모든 수신 주문이 포함됩니다.)
- 다시 시도 (여기에는 실패한 주문에 대한 재시도 문서가 포함됩니다.)

버킷 만들기 '메타데이터'. 기본 범위와 기본 컬렉션을 사용하겠습니다. 메타데이터 버킷은 이벤트 메타데이터에 사용됩니다.

에 인덱스를 생성합니다. 다시 시도 컬렉션을 생성합니다. 재시도 수신기는 N1QL을 사용하여 컬렉션에 포함된 모든 문서를 쿼리하므로 쿼리가 실행되려면 인덱스가 있어야 합니다.

CREATE PRIMARY INDEX idx_default_primary ON orders._default.retry USING GSI;

1	만들기 기본 INDEX IDX_기본_기본 켜기 주문._기본값.다시 시도 사용 GSI;

데이터 모델 주문 문서

이 샘플 애플리케이션에서는 주문 문서에 관련 필드만 포함하는 경량 데이터 모델을 사용합니다. 주문 문서에서 일반적으로 기대할 수 있는 다른 많은 필드는 생략되었습니다.

{
  "email": "customer_email",
  "paymentStatus": "initiated",
  "confirmationEmailSent": false,
  "items": [
    {
      "name": "Swedish Meatballs 500g",
      "amount": 2,
      "unitPrice": 9.95
    }
  ]
}

{

"이메일": "customer_email",

"paymentStatus": "시작됨",

"확인 이메일 전송": false,

"items": [

{

"name": "스웨덴 미트볼 500g",

"금액": 2,

"unitPrice": 9.95

}

]

}

데이터 모델 재시도 문서

재시도 문서에는 주문 문서의 문서 ID, 시도 카운터 및 타임스탬프와 같은 몇 가지 기본 속성이 포함되어 있습니다. 재시도 문서의 유형 속성은 애플리케이션에서 필요하지 않지만, 애플리케이션이 배송 및 배달 업데이트도 전송하도록 확장되는 경우 이메일 알림 유형을 결정하는 데 유용할 수 있습니다.

{
  "type": "confirmation",
  "docId": "order_140",
  "attempt": 1,
  "ts": 1632775908319
}

{

"type": "확인",

"docId": "order_140",

"시도": 1,

"ts": 1632775908319

}

이메일 서비스 모의

로컬 웹 서버를 실행하는 간단한 Python 스크립트를 사용하여 이메일 서비스를 모의해 보겠습니다. 이 스크립트는 무작위로 HTTP 200 OK로 응답하거나 HTTP 406으로 응답하여 실패를 표시합니다.

31번 줄의 로컬 컴퓨터의 IP 주소로 IP 주소를 업데이트합니다.server = ThreadedHTTPServer(('IP로 대체', 9080), 핸들러)
다음을 실행하여 스크립트를 시작하세요.

이벤트 리스너

이제 모든 준비가 완료되었으므로 두 개의 이벤트 리스너를 추가할 수 있습니다:

EVT_SEND_CONFIRM_EMAIL - 이메일 서비스와의 통합을 제공합니다.
EVT_SEND_확인_이메일_재시도 - 에는 재시도 로직이 포함되어 있습니다.

청취자는 여기에서 확인할 수 있습니다: https://github.com/puhhma/cb_eventing_retry_sample
리스너 가져오기(json 파일)을 카우치베이스 이벤트 서비스에 추가합니다.

리스너가 작동하려면 이 문서에 사용된 명명 규칙을 따라야 합니다.

검토 EVT_SEND_CONFIRM_EMAIL 리스너

이벤트 리스너 구성:

이벤트 리스너가 인바운드 컬렉션의 주문 버킷.
그리고 메타데이터 버킷은 리스너 메타데이터를 저장하는 데 사용됩니다.
버킷 별칭 bkt_order_inbound 그리고 bkt_order_retry 를 참조하여 해당 인바운드 그리고 다시 시도 컬렉션의 주문 버킷
그리고 curlEmail서비스호스트 는 모의 이메일 서비스에 대한 URL 별칭을 지정합니다. IP 주소로 업데이트해야 합니다.

// OnUpdate is invoked for all documents created/updated in the 'inbound' bucket
function OnUpdate(doc, meta) {
    // determine if document status is 'paid' & confirmation email was not previously sent
    if( doc.paymentStatus === "paid" && !doc.confirmationEmailSent ) {
        SendConfirmationMail(doc, meta.id);
    } else {
      if (debug_level > 1)
        log("Nothing to do for: " + meta.id);
    }
}

function SendConfirmationMail(doc, docId) {
  try {
      // build the request to the EmailService
      var request = {
          path: 'sendConfirmation',
          body: doc
      };
      //  perform the cURL request using the URL alias 'curlEmailServiceHost' from the settings
      var response = curl('POST', curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log("docId", docId, "cURL POST failed response.status:",response.status);
          }
          // create retry document referencing the documentId and store in 'retry' bucket
          bkt_order_retry[docId] = {
              "docId": docId,
              "attempt": 1,
              "ts": Date.now()
          }
      } else {
          if (debug_level > 5) {
              log("cURL POST success, sent",docId,"response.body:",response.body);
          }
          // update confirmationEmailSent status
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
      }
  } catch (e) {
      log("ERROR cURL request had an exception:",e)
  }
}

// '인바운드' 버킷에서 생성/업데이트된 모든 문서에 대해 OnUpdate가 호출됩니다.

함수 온업데이트(doc, 메타) {

// 문서 상태가 '결제 완료'이고 확인 이메일이 이전에 전송되지 않았는지 확인합니다.

만약( doc.결제 상태 === "유료" && !doc.확인이메일 전송 ) {

확인 메일 보내기(doc, 메타.id);

} else {

만약 (debug_level > 1)

로그("할 일이 없습니다: " + 메타.id);

}

함수 확인 메일 보내기(doc, docId) {

시도 {

// 이메일 서비스에 요청을 빌드합니다.

var 요청 = {

경로: 'sendConfirmation',

body: doc

};

// 설정에서 URL 별칭 'curlEmailServiceHost'를 사용하여 cURL 요청을 수행합니다.

var 응답 = curl('POST', curlEmail서비스호스트, 요청);

만약 (응답.상태 != 200) {

// 예상대로 작동하지 않습니다.

만약 (debug_level > 1) {

로그("docId", docId, "cURL POST 실패 응답.상태:",응답.상태);

}

// documentId를 참조하여 재시도 문서를 생성하고 '재시도' 버킷에 저장합니다.

bkt_order_retry[docId] = {

"docId": docId,

"시도": 1,

"ts": 날짜.지금()

}

} else {

만약 (debug_level > 5) {

로그("cURL POST 성공, 전송됨",docId,"response.body:",응답.body);

}

// 확인 이메일 전송 상태 업데이트

doc.확인이메일 전송 = true;

bkt_order_inbound[docId] = doc;

}

} catch (e) {

로그("오류 cURL 요청에 예외가 발생했습니다:",e)

}

자세한 내용은 인라인 댓글을 참조하세요.
그리고 온업데이트 함수는 주문 문서가 업데이트되거나 생성되면 트리거됩니다.
요청이 구성되고 cURL을 사용하여 HTTP POST 요청이 이메일 서비스로 전송됩니다.
결과가 평가됩니다. HTTP 응답이 성공하지 못한 경우 재시도 문서가 작성되어 재시도 문서에 추가됩니다. 다시 시도 컬렉션.

이벤트 리스너가 다시 시도 컬렉션의 주문 버킷.
그리고 메타데이터 버킷은 리스너 메타데이터를 저장하는 데 사용됩니다.
버킷 별칭 bkt_order_inbound 그리고 bkt_order_retry 를 참조하여 해당 인바운드 그리고 다시 시도 주문 버킷의 컬렉션
그리고 curlEmail서비스호스트 모의 이메일 서비스에 대한 URL 별칭을 지정합니다.
그리고 재시도 타이머 간격 는 타이머 간격을 초 단위로 지정합니다.

function OnUpdate(doc, meta) {
    if (meta.id === "allow_retrys") {
        // the timer is initialized by creating document with id = 'allow_retrys'
        CreateRetryTimer({"id": meta.id, "mode": "initial"});
    } else if (doc.fireRetry) {
      // process retry documents
      SendConfirmationMail(doc, meta.id);
    }
}

function CreateRetryTimer(context) {
    if (debug_level > 2) {
        log('From CreateRetryTimer: creating timer', context.mode, context.id);
    }
    // Create a timestamp 'retryTimerInterval' seconds (from the settings) from now
    var timerStartTime = new Date();
    // Get current time & add 'retryTimerInterval' sec. to it.
    timerStartTime.setSeconds(timerStartTime.getSeconds() + retryTimerInterval);
    // Create a document to use as out for our context
    createTimer(RetryTimerCallback, timerStartTime, context.id, context);
}

function RetryTimerCallback(context) {
    if (debug_level > 2) {
        log('From RetryTimerCallback: timer fired', context);
    }
    // rearm the timer ASAP, to ensure timer keeps running in the event
    // of later  errors or script timeouts in later "recurring work".
    CreateRetryTimer({ "id": context.id, "mode": "via_callback" });
    // Update all retry documents in the 'retry' bucket. Exclude the 'allow_retys' document
    // and any documents that were created more than 15 seconds ago, in order to avoid retry 'to early'.
    N1QL("UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != 'allow_retrys' AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, 'second')");
}

function SendConfirmationMail(retryDoc, docId) {
  try {
      // resolve order document by id
      var doc = bkt_order_inbound[docId];
      // build the request
      var request = {
          path: 'sendConfirmation',
          body: doc
      };
      //  perform the cURL request using the URL alias from the settings
      var response = curl('POST', curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log("docId", docId, "cURL POST failed response.status:",response.status);
          }
        // increment attempt count in retry document
        retryDoc.attempt = ++retryDoc.attempt;
        // Set fireRetry = false, to avoid retry execution with this document change
        retryDoc.fireRetry = false;
        retryDoc.ts = Date.now();
        // update retry document
        bkt_order_retry[docId] = retryDoc;
      } else {
          if (debug_level > 5) {
              log("cURL POST success, sent",docId,"response.body:",response.body);
          }
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
          // delete the retry document
          delete bkt_order_retry[docId];
      }  } catch (e) {
      log("ERROR cURL request had an exception:",e)
  }
}

함수 온업데이트(doc, 메타) {

만약 (메타.id === "허용_재시도") {

// 타이머는 id = 'allow_retries'로 문서를 생성하여 초기화됩니다.

레티 타이머 생성({"id": 메타.id, "모드": "initial"});

} else 만약 (doc.fireRetry) {

// 문서 재시도 처리

확인 메일 보내기(doc, 메타.id);

}

함수 레티 타이머 생성(컨텍스트) {

만약 (debug_level > 2) {

로그('From CreateRetryTimer: 타이머 생성', 컨텍스트.모드, 컨텍스트.id);

}

// 지금부터 (설정에서) 타임스탬프 'retryTimerInterval' 초를 생성합니다.

var 타이머 시작 시간 = new 날짜();

// 현재 시간을 가져와서 'retryTimerInterval' 초를 추가합니다.

타이머 시작 시간.setSeconds(타이머 시작 시간.getSeconds() + 재시도 타이머 간격);

// 컨텍스트에 맞게 아웃으로 사용할 문서를 만듭니다.

createTimer(RetryTimerCallback, 타이머 시작 시간, 컨텍스트.id, 컨텍스트);

}

함수 RetryTimerCallback(컨텍스트) {

만약 (debug_level > 2) {

로그('RetryTimerCallback에서: 타이머가 실행됨', 컨텍스트);

}

// 타이머를 최대한 빨리 재장전하여 타이머가 이벤트에서 계속 실행되도록 합니다.

// 이후 '반복 작업'에서 나중에 발생하는 오류나 스크립트 시간 초과를 방지합니다.

레티 타이머 생성({ "id": 컨텍스트.id, "모드": "via_callback" });

// '재시도' 버킷에 있는 모든 재시도 문서를 업데이트합니다. 'allow_retys' 문서 제외하기

// 및 15초 이상 전에 생성된 모든 문서를 '일찍' 다시 시도하지 않도록 합니다.

N1QL("UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != 'allow_retrys' AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, 'second')");

}

함수 확인 메일 보내기(retryDoc, docId) {

시도 {

// ID로 주문 문서 확인

var doc = bkt_order_inbound[docId];

// 요청 빌드

var 요청 = {

경로: 'sendConfirmation',

body: doc

};

// 설정에서 URL 별칭을 사용하여 cURL 요청을 수행합니다.

var 응답 = curl('POST', curlEmail서비스호스트, 요청);

만약 (응답.상태 != 200) {

// 예상대로 작동하지 않습니다.

만약 (debug_level > 1) {

로그("docId", docId, "cURL POST 실패 응답.상태:",응답.상태);

}

// 재시도 문서에서 시도 횟수 증가

retryDoc.시도 = ++retryDoc.시도;

// 이 문서 변경으로 재시도 실행을 방지하려면 fireRetry = false로 설정합니다.

retryDoc.fireRetry = false;

retryDoc.ts = 날짜.지금();

// 재시도 문서 업데이트

bkt_order_retry[docId] = retryDoc;

} else {

만약 (debug_level > 5) {

로그("cURL POST 성공, 전송됨",docId,"response.body:",응답.body);

}

doc.확인이메일 전송 = true;

bkt_order_inbound[docId] = doc;

// 재시도 문서 삭제

삭제 bkt_order_retry[docId];

} } catch (e) {

로그("오류 cURL 요청에 예외가 발생했습니다:",e)

}

타이머는 ID가 다음과 같은 문서를 추가하여 시작됩니다. 허용_재시도 를 다시 시도 컬렉션
그런 다음 타이머가 초기화되고 타이머와 연결된 RetryTimerCallback 함수가 호출됩니다.
타이머가 실행되면 RetryTimerCallback 함수가 호출됩니다.
재시도 메커니즘을 진행하기 전에 첫 번째 단계로 새 타이머가 생성되어 나중에 오류가 발생할 경우 계속 실행되도록 합니다.
N1QL 쿼리는 모든 업데이트에 사용됩니다. 다시 시도 문서에 있는 다시 시도 컬렉션을 추가하여 fireRetry 속성을 문서에 추가합니다.
문서가 변경될 때마다 문서 업데이트 이벤트가 발생하고 재시도 메커니즘이 실행됩니다.
주문 문서는 인바운드 컬렉션을 수집하고 이메일 서비스는 cURL을 통해 호출됩니다.
실패 시 다시 시도 문서가 업데이트되고 시도 카운터 증가

샘플 애플리케이션 테스트

이제 마지막으로 샘플 애플리케이션을 테스트할 차례입니다:

모의 이메일 서비스가 실행 중인지 확인합니다.
시작하기 EVT_SEND_CONFIRM_EMAIL 이벤트 리스너를 사용하되 EVT_SEND_확인_이메일_재시도 리스너가 현재 중지되었습니다.
Couchbase 콘솔에서 샘플 주문 문서를 만듭니다(위의 데이터 모델 참조).
성공 응답의 경우 확인이메일 전송 속성이 참으로 업데이트됩니다. 주문 문서.
실패할 경우 재시도 문서는 다시 시도 수집. 이메일 서비스에서 무작위로 오류를 회신하므로 오류가 발생할 때까지 #3 단계를 반복하세요.
이제 오류를 캡처했으니 재시도 이벤트 리스너를 시작하겠습니다. EVT_SEND_확인_이메일_재시도
아이디가 'allow_retries'인 문서를 만듭니다. 이렇게 하면 재시도 메커니즘이 초기화됩니다.
잠시 후 리스너가 활성화되고 문서 처리가 시작됩니다. 다시 시도 컬렉션.
이메일 서비스 업데이트에 실패할 때마다 '시도' 속성이 증가한다는 점을 유의하세요. 성공할 경우 주문 문서가 업데이트되고 그에 해당하는 다시 시도 문서에서 제거된 다시 시도 컬렉션.

이메일 서비스 모의 테스트의 응답은 무작위이므로 예상되는 동작을 관찰하기 위해 위의 단계를 반복해야 할 수도 있습니다.

결론

이 글에서는 카우치베이스 이벤트와 외부 REST 서비스를 통합할 때 오류 조건을 처리하는 재시도 메커니즘에 대해 간략하게 설명합니다. 이 솔루션 또는 이와 유사한 솔루션을 사용하면 외부 서비스가 일시적으로 오작동하는 경우에도 예상된 작업이 수행되도록 보장할 수 있습니다.

재시도 메커니즘을 고려할 때는 재시도 횟수, Couchbase Eventing에 사용 가능한 워커 스레드 및 외부 서비스에서 처리할 수 있는 요청 등 다양한 요소를 고려해야 합니다.

카우치베이스 이벤트의 내부에 대한 자세한 내용은 여기에서 확인할 수 있습니다: https://docs.couchbase.com/server/current/eventing/eventing-overview.html

이 글에 대한 기술적 인사이트와 지원을 제공해 주신 존 스트라발라(Couchbase의 수석 제품 관리자)에게 감사드립니다.

마리안 풀, 솔루션 엔지니어

이 문서 공유하기

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

카우치베이스 이벤트 처리 오류 및 재시도

받은 편지함에서 카우치베이스 블로그 업데이트 받기

작성자

게시자 마리안 풀, 솔루션 엔지니어

댓글 남기기 응답 취소

카우치베이스 카펠라를 시작할 준비가 되셨나요?

구축 시작

카펠라 무료 사용

연락하기