learn more

카프카가 자꾸 말썽을 부리고 있다.

왜일까...

공식 홈페이지의 기본 개념을 살짝 정리해보자.

https://kafka.apache.org/090/documentation.html#gettingStarted

잘못된 정보 수정은 환영입니다.

​

카프카는 topic을 통해 메시지 피드를 유지 관리한다.

topic producer를 통해 kafka cluster에 메시지를 적재한다.

consumer를 통해 kafka cluster에서 topic을 감지하고 메시지를 가져온다.

그림의 kafka cluster는 1개지만 그 이상의 서버로 구성될 수 있다.

​

cluster에 1개의 토픽에 관한 파티션을 살펴보면 위의 그림과 같다.

파티션을 1개로 놓는것도 가능하지만,
나눠 놓는것으로 메시지의 read와 write의 부하 분산을 줄여 성능 개선을 할 수 있다.
​

각 파티션의 메시지는 지속적으로 추가되고, 정렬 또는 변경이 불가능하다.

파티션 별로 메시지에 오프셋이라는 순차 ID가 부여된다.

​

메시지 유지 기간

cluster는 구성 가능 기간 동안 소비 여부에 관계없이 모든 메시지를 유지한다.

(보존 기간이 2일로 설정된 경우 메시지가 게시된 후 2일 동안 사용할 수 있고, 그 후에는 공간 확보를 위해 삭제한다.)

​

offset

각각의 consumer는 오프셋 번호를 가지고 있다.

오프셋은 소비자 마음대로 왔다 갔다 할 수 있으며,

그로 인해 처리했던 메시지를 다시 처리할 수 있다.

consumer의 오프셋 변화는 cluster 또는 producer에 영향을 미치지 않는다.

​

파티션

파티션은 1개의 topic에 대해 많은 양의 파티션으로 구성할 수 있고, 이것은 병렬 처리가 가능하다.

주의할 점은

kafka를 사용하면서 파티션을 늘릴 수는 있지만 줄일 수는 없다.

파티션을 줄이고 싶다면 토픽을 삭제하고 다시 만들어야 한다.

​

복사 전략

replication-factor 옵션에 따라 여러 cluster 서버에 데이터를 복제하여 안전하게 관리할 수 있다.

2로 설정한다면 원본과 복사본의 갯수를 합쳐서 2개의 데이터가 존재한다.

전략은 leader - follwers 방식으로 아래 사진을 참고할 수 있다.

leader-follwers 방식

클러스터하나의 클러스터 안에는 여러개의 브로커가 존재한다.
각 파티션에서 리더를 선출하여 팔로워들은 리더의 데이터를 복사한다.
리더가 죽게 되면 팔로워 중 하나를 리더로 선출한다.

앞에서 파티션을 늘려 부하 분산을 할 수 있다고 언급했는데,

무분별한 파티션 증설은 안 좋은 영향을 줄 수 있다.

이유는 위의 리더 선출과 복사 전략 때문인데,만약 Brooker101 이 다운된다면 Broker 102의 partition 0 은 일시적으로 사용할 수 없게 된다.
(새로운 리더 선출로부터 영향)

또한 다운된 브로커가 컨트롤러 역할을 수행하고 있었다면, 새로운 컨트롤러 선출을 위해서 모든 파티션의 데이터를 읽어야하기 때문에 더 큰 지연을 불러올 수 있다.

​

메시지 적제

producer는 topic 내의 어떤 파티션에 메시지를 할당할지 선택해야 한다.

방식은 메시지의 key 기반으로 분할되거나,(따로 공부하자)

key 값이 없다면 기본적으로 라운드 로빈 방식으로 분할된다.

메시지를 컨슘 할 때, 파티션 내부에서는 순서를 보장한다.

하지만 토픽 단위로 봤을 때 라운드로빈 기반으로 메시지를 저장했다면,

각각의 파티션끼리는 순서를 보장하지 않으니 주의하자.

​

메시지 소비

kafka는 소비자 그룹을 제공한다.

consumer에 group-id를 지정한다.

(C1과 C2는 A라는 group-id를 지정하여 하나의 그룹으로 본다.)

동일한 group-id를 가진 consumer 중 하나라도 메시지를 소비하면,

같은 그룹 내의 consumer들은 메시지를 소비하지 않고 넘어간다.

(C1과 C2는 1개의 topic에 대한 파티션처럼 동작한다)

Kafka

오픈 소스 분산 이벤트 스트리밍 플랫폼입니다. 

kafka는 실시간으로 데이터 스트림을 수집하여 처리하기 위한 플랫폼으로, 메시지 스트리밍 데이터를 쓰고 읽게 해주는 게시(pub)-구독(sub)형 메시징 시스템입니다. 

Apache Kafka

동작 원리

카프카는 3가지 컴포넌트로 구분됩니다.

Producer : kafka에 데이터 입력 클라이언트.
Broker : 중간 역할을 해줄 클러스터.
Consumer : kafka의 데이터를 가져오는 클라이언트.

producer는 이벤트(데이터, record, message)를 게시(쓰기)하는 클라이언트 애플리케이션이고,
consumer는 이벤트를 읽는 애플리케이션입니다.

브로커라는 클러스터에 파일 시스템의 폴더와 같은 ‘topic’을 생성하고,
파일처럼 이벤트를 생성하고 읽는 것이죠.

하지만 하나의 topic에 많은 producer와 consumer가 접근하게 되면, 부하가 걸리고 딜레이가 생길 수 있는데, 이러한 것들 때문에 topic은 n개의 partition으로 분산됩니다.

새 이벤트가 topic에 게시되면 실제로는 topic의 partition 중의 1개에 추가됩니다.

각각의 파티션은 append-only 방식으로 쌓여가는 로그 파일 형태입니다.

offset이라고 불리는 각각의 파티션 내 위치를 의미하는 식별자 정보를 가지고 이벤트를 쓰고 읽을 수 있습니다. 한번 쓰여진 이벤트는 영구적으로 저장됩니다(영속성 보장). 타 메세지 시스템과는 다르게, 읽는다고 삭제되는게 아니라 topic별 구성 설정을 통해 이벤트를 유지하는 기간을 정의하여 삭제 관리합니다.

 

각각의 파티션 내에선 순서가 보장되고, topic 내의 파티션 간의 순서는 보장되지 않습니다.

만약, 특정 이벤트들간의 작동 순서대로 읽고 싶다면, 이벤트 키(파티션 키)를 통해 동일한 파티션에 기록하게 하여 순서를 보장해야합니다.(없으면 라운드로빈 방식으로 파티션을 돌아가며 씀)

Consumer

컨슈머 그룹은 컨슈머들을 논리적으로 묶은 것입니다. 

컨슈머 그룹 내 리더 컨슈머가 해당 메시지를 누가 읽을건지 선정하고 나머지 컨슈머는 해당 메시지를 읽지 않습니다.

카프카의 특징

1) 카프카는 브로커라 불리는 하나 또는 그 이상의 서버들로 이루어진 클러스터로, 토픽의 파티션들을 여러 브로커에 배포합니다.
이로 인해, 하나의 토픽을 여러 consumer에 의해 동시 처리가 가능하고, 서로 다른 브로커에 있는 파티션에 접속함으로써 메시지 처리량을 높일 수 있습니다.

2) 카프카는 복사정책으로 여러 브로커에 동일 파티션의 복사본(replica)을 3개 이상 유지합니다.
이를 통해 1개의 브로커가 장애가 발생해도 복사본을 통해 지속적으로 메시지 제공이 가능합니다.

3) 카프카는 컨슈머 리더쪽에서 현재 구독중인 파티션 할당 정책을 담당합니다.
broker에서 관리안하고 컨슈머쪽에서 관리하는 이유는  broker를 재시작할 필요 없게 하여 유연성과 확장성을 가지게 됩니다.

Spring Security의 filter chain에 있는 필터들을 알아보자!

Spring Security의 기본필터를 알아보는 시리즈

바로 들어가자 - doFilter

ChannelProcessingFilter 의 doFilter 부분을 가지고 왔다.

@Override
public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain)
      throws IOException, ServletException {
   HttpServletRequest request = (HttpServletRequest) req;
   HttpServletResponse response = (HttpServletResponse) res;
   FilterInvocation filterInvocation = new FilterInvocation(request, response, chain);
   Collection<ConfigAttribute> attributes = this.securityMetadataSource.getAttributes(filterInvocation);
   if (attributes != null) {
      this.logger.debug(LogMessage.format("Request: %s; ConfigAttributes: %s", filterInvocation, attributes));
      this.channelDecisionManager.decide(filterInvocation, attributes);
      if (filterInvocation.getResponse().isCommitted()) {
         return;
      }
   }
   chain.doFilter(request, response);
}

여기서 핵심은 channelDecisionManager.decide 이다.

this.channelDecisionManager.decide(filterInvocation, attributes);

ChannelDecisionManager 의 decide 의 구현체를 보면,

ANY_CHANNEL이라는 attribute가 포함되어 있으면 바로 리턴을 해주고,

ChannelProcessor가 지정되어 있으면 decide 메소드로 연결되어 있다.

ChannelProcessor가 무엇인가하면 프로토콜에 따라 어떻게 요청을 처리할 지 각 채널 (http, https 등) 처리를 관리하는 프로세서이다.

ChannelProcessor의 구현체는 2개 있는데 Spring으로 주로 web 요청을 처리하기 때문에,

InsecureChannelProcessor는 http를, SecureChannelProcessor는 https를 처리하는 걸로 알고 있으면 된다

조금 더 자세하게는 각 프로세스 안에서는 AbstractRetryEntryPoint의 commence라는 동작으로 연결되는데, 

port와 request 형태, EntryPoint를 잡아주는 동작을 하고 있고

AbstractRetryEntryPoint 구현체는 RetryWithHttpEntryPoint, RetryWithHttpsEntryPoint 2개 있다.

RetryWithHttpsEntryPoint는 이렇게 생겼다. 

실은 조사한 건 여기까지인데,

아쉬우니 chatGPT의 설명을 끝으로 마치려한다.

 ChannelProcessingFilter는 웹 애플리케이션에 대한 요청의 인증 및 승인을 처리하는 데 사용되는 Spring Security 프레임워크의 클래스입니다.

이 필터는 들어오는 요청을 검사하고 어떤 보안 속성이 적용되는지 결정한 다음 인증 및 권한 부여 프로세스를 Spring Security 프레임워크 내의 적절한 구성 요소에 위임하는 역할을 합니다.

ChannelProcessingFilter는 HTTPS와 같은 보안 채널이 필요한 요청을 처리하도록 특별히 설계되었으며 필요한 경우 요청을 보안 채널로 리디렉션할 수 있습니다. 일반적으로 AuthenticationFilter 및 AuthorizationFilter와 같은 다른 보안 필터보다 먼저 필터 체인의 초기에 배치됩니다.

전반적으로 ChannelProcessingFilter는 웹 애플리케이션에 대한 요청이 적절하게 인증 및 승인되고 민감한 정보가 네트워크를 통해 안전하게 전송되도록 하는 데 도움이 되는 Spring Security 프레임워크의 중요한 구성 요소입니다.

Bye!

스프링 시큐리티를 내 스프링 프로젝트에서 사용하고 있다면, 모두들 체감은 안되지만 저절로 사용되고 있는 기본 필터를 알아보자 !

---

Spring Security Architecture

• Spring Security는 애플리케이션의 보안에 핵심적인 "인증"과 "인가"를 Filter와 annotation을 사용해서 손쉽게 구현할 수 있도록 도와준다.

AuthenticationManager : 사용자 인증 관련 처리
AccessDecisionManager : 사용자가 보호받는 리소스에 접근할 수 있는 적절한 권한이 있는지 확인 

FilterChainProxy

Spring Security는 서블릿 필터 javax.servlet.Filter의 구현체로 동작한다.
@EnableWebSecurity 어노테이션과 함께 WebSecurityConfigurerAdapter 추상 클래스를 상속받아 구현함
웹 요청은 이러한 필터 체인을 차례로 통과하게 됨 보통 SpringSecurityFilterChain 이라는 이름으로 Bean 등록

 

필터 체인에 Custom Filter를 추가해서 쓰는 일도 많지만 기본적으로 설정하지 않아도 작동되는 많은 필터들에 대해 오늘 알아보려 한다.

스프링 공식 문서에서 안내하는 기본 필터는 아래와 같다.

영어 주의

Filter Ordering

The order that filters are defined in the chain is very important. Irrespective of which filters you are actually using, the order should be as follows:

물론 Spring Security의 `FilterOrderRegistration` 클래스를 보면 실제 spring security에서 사용하는 Filter Chain을 확인할 수 있다.

- FilterOrderRegistration에 등록된 필터 체인 - 코드로 보기

FilterOrderRegistration() {
   Step order = new Step(INITIAL_ORDER, ORDER_STEP);
   put(ChannelProcessingFilter.class, order.next());
   order.next(); // gh-8105
   put(WebAsyncManagerIntegrationFilter.class, order.next());
   put(SecurityContextPersistenceFilter.class, order.next());
   put(HeaderWriterFilter.class, order.next());
   put(CorsFilter.class, order.next());
   put(CsrfFilter.class, order.next());
   put(LogoutFilter.class, order.next());
   this.filterToOrder.put(
         "org.springframework.security.oauth2.client.web.OAuth2AuthorizationRequestRedirectFilter",
         order.next());
   this.filterToOrder.put(
         "org.springframework.security.saml2.provider.service.servlet.filter.Saml2WebSsoAuthenticationRequestFilter",
         order.next());
   put(X509AuthenticationFilter.class, order.next());
   put(AbstractPreAuthenticatedProcessingFilter.class, order.next());
   this.filterToOrder.put("org.springframework.security.cas.web.CasAuthenticationFilter", order.next());
   this.filterToOrder.put("org.springframework.security.oauth2.client.web.OAuth2LoginAuthenticationFilter",
         order.next());
   this.filterToOrder.put(
         "org.springframework.security.saml2.provider.service.servlet.filter.Saml2WebSsoAuthenticationFilter",
         order.next());
   put(UsernamePasswordAuthenticationFilter.class, order.next());
   order.next(); // gh-8105
   this.filterToOrder.put("org.springframework.security.openid.OpenIDAuthenticationFilter", order.next());
   put(DefaultLoginPageGeneratingFilter.class, order.next());
   put(DefaultLogoutPageGeneratingFilter.class, order.next());
   put(ConcurrentSessionFilter.class, order.next());
   put(DigestAuthenticationFilter.class, order.next());
   this.filterToOrder.put(
         "org.springframework.security.oauth2.server.resource.web.BearerTokenAuthenticationFilter",
         order.next());
   put(BasicAuthenticationFilter.class, order.next());
   put(RequestCacheAwareFilter.class, order.next());
   put(SecurityContextHolderAwareRequestFilter.class, order.next());
   put(JaasApiIntegrationFilter.class, order.next());
   put(RememberMeAuthenticationFilter.class, order.next());
   put(AnonymousAuthenticationFilter.class, order.next());
   this.filterToOrder.put("org.springframework.security.oauth2.client.web.OAuth2AuthorizationCodeGrantFilter",
         order.next());
   put(SessionManagementFilter.class, order.next());
   put(ExceptionTranslationFilter.class, order.next());
   put(FilterSecurityInterceptor.class, order.next());
   put(AuthorizationFilter.class, order.next());
   put(SwitchUserFilter.class, order.next());
}

여튼 그렇다. 공식 문서에 나온 것부터 하나씩 알아보자

필터들

1. ChannelProcessingFilter 🔗

2. SecurityContextPersistenceFilter

3. ConcurrentSessionFilter

4. UsernamePasswordAuthenticationFilter

5. SecurityContextHolderAwareRequestFilter

6. RememberMeAuthenticationFilter

7. AnonymousAuthenticationFilter

8. ExceptionTranslationFilter

9. FilterSecurityInterceptor

참고

- https://docs.spring.io/spring-security/reference/index.html

- https://hanjo8813.github.io/til/42/

고객에 따른 분류

플랫폼 (서로 다른 롤의 B2C, B2B가 혼합되어 거래활동이 일어나는 형태), B2C, B2B

프로덕트가 제공되는 형태에 따라 분류

https://brainhub.eu/library/cloud-architecture-saas-faas-xaas

https://m.blog.naver.com/PostView.naver?isHttpsRedirect=true&blogId=doonibest&logNo=220861308003 

그리고.

정보 제공을 위한 소프트웨어, 사람과 사람을 연결하는 형태의 소프트웨어

기업들의 MSA 이야기

MSA 구조: 구글 vs 넷플릭스

https://netmarble.engineering/past-1year-to-be-server-programmer-as-one-developer/?fbclid=IwAR1U93vf2aQE_XB-bquMYKST4RyOP7_OVNjc5v35U01wmt0aSeSBzNKTVhg

나도 나아가겠다

AOP 는 Aspect-Oriented Programming 의 약자입니다. AOP 로 로직상에서 문제를 해결하고자 하는 핵심 기능과 공통 기능을 분리하는 프로그래밍 방법입니다. Spring의 관점지향 프로그래밍은 Bean, Filter를 이용해 구현 되어 있습니다. 따라서 Spring web 서버를 실행하면서 혹은 Bean에 등록된 메소드들이 실행하기 전 후에 일관된 동작을 추가해줄 수 있습니다. 로깅이나 성능 측정, Validation 검사, 인증 검사 등에 쓰입니다.

틈새 원리 탐구

Spring AOP는 타겟이 되는 Bean 객체를 JDK 다이나믹 프록시 형태로 감싸 Spring 동작시 전 후로 동작을 수행합니다. 혹은 Bean의 인스턴스화 형태에 따라 CGLIB로도 AOP를 구현하기도 합니다.

AspectJ는 모든 객체에 적용 가능하지만 적용하기 굉장히 까다롭고, Spring AOP는 개발자가 aspect class 하나로 적용하도록 만든 것이라 생각됩니다. AspectJ를 Spring이 관리하는 Bean에만 적용가능하게 만들어져 있는 형태입니다.

스프링 공식 문서의 AOP 원리 설명

AOP 의 주요 키워드

Target

공통 기능을 부여할 대상입니다. 즉, 핵심 기능을 담당하는 비즈니스 로직이고, 어떤 관심사들과도 관계를 맺지 않습니다.

Aspect

여러 핵심 기능에 적용될 관심사 모듈입니다. Aspect 에는 구체적인 기능을 구현한 Advice 와 Advice 가 어디에서 적용될지를 결정하는 PointCut 을 포함하고 있습니다.

Advice

Advice 는 공통 기능을 담은 구현체입니다. Advice 는 Aspect 가 무엇을 언제 적용할 지를 정의합니다.

PointCut

PointCut 은 공통 기능이 적용될 대상을 결정합니다.

JoinPoint

Advice 가 적용될 지점을 의미합니다. Spring 에서는 메서드만이 JoinPoint 로 제공하고 있습니다.

Proxy

Target 을 랩핑하여 Target 의 요청을 대신 받아주는 오브젝트입니다.

Weaving

지정된 객채에 Aspect 를 적용해서 새로운 프록시 객체를 생성하여 적용하는 것을 말합니다. 즉, 공통 코드를 핵심 로직 코드에 삽입하는 과정입니다.

Advice 정리

@Before

Advice 의 Target 메서드가 호출되기 전에 실행되는 Advice 입니다.

@After

Advice 의 Target 메서드가 호출된 후에 그 결과에 상관없이(성공 or 실패) 실행되는 Advice 입니다.

@AfterReturning

Target 메서드가 정상적으로 실행된 후에 실행되는 Advice 입니다.

@AfterThrowingAdvice

Target 메서드가 수행도중 예외가 발생했을 때 예외를 던지게 되면 실행되는 Advice 입니다.

@Around

Advice 가 Target 메서드를 감싸서 Target 메서드 호출 전과 후에 실행되는 Advice 입니다. @Around 는 ProceedingJoinPoint 의 proceed() 를 수행하는데, proceed() 의 리턴 값이 sub 메서드의 리턴 값이다. 즉, @Advice 내부에서 결과값을 제어할 수 있다.

PointCut 표현식

포인트 컷에는 아래와 같은 명시자를 이용할 수 있습니다.

• execution : Advice 를 적용할 메서드를 명시할 때 사용합니다.
• within : 특정 타입에 속하는 메서드를 JoinPoint 로 설정되도록 명시할 때 사용합니다.
• bean : 스프링 버전 2.5 버전부터 지원하기 시작했으며, 스프링 빈을 이용하여 JoinPoint 를 설정합니다.

execution 명시자

execution 명시자

execution([수식어] 리턴 타입 [클래스 이름].이름(파라미터))

• 수식어 : public, private 등 수식어를 명시합니다. (생략가능)
• 리턴 타입 : 리턴 타입을 명시합니다.
• 클래스 이름 및 이름 : 클래스 이름과 메서드 이름을 명시합니다. (클래스 이름은 풀 패키지 명으로 명시해야합니다. 클래스 이름은 생략할 수도 있습니다.)
• 파라미터 : 메서드의 파라미터를 명시합니다.
• " * " : 모든 값을 표현합니다.
• " .. " : 0개 이상을 의미합니다.

ex)

execution(public Integer com.edu.aop..(*))

• com.edu.aop 패키지에 속해있고, 파라미터가 1개인 모든 메서드

execution(* com.edu...get(..))

• com.edu 패키지 및 하위 패키지에 속해있고, 이름이 get 으로 시작하는 파라미터가 0개 이상인 모든 메서드

execution(* com.edu.aop..Service.(..))

• com.edu.aop 패키지 및 하위 패키지에 속해있고, 이름이 Service 로 끝나는 클래스의 파라미터가 0개 이상인 모든 메서드

execution(* com.edu.aop.BoardService.*(..))

• com.edu.aop.BoardService 클래스에 속한 파라미터가 0개 이상인 모든 메서드

execution(* some(*, *))

• 메서드 이름이 some 으로 시작하고 파라미터가 2개인 모든 메서드

within 명시자

ex)

within(com.edu.aop.SomeService)

• com.edu.aop.SomeService 클래스의 모든 메서드

within(com.edu.aop.*)

• com.edu.aop 패키지의 모든 메서드

within(com.edu.aop..*)

• com.edu.aop 패키지 및 하위 패키지의 모든 메서드

bean 명시자

ex)

bean(someBean)

• 이름이 someBean인 빈의 모든 메서드

bean(somm*)

• 빈의 이름이 some 으로 시작하는 빈의 모든 메서드

아래의 @PointCut 애너테이션을 이용하여 JoinPoint 를 지정할 수 있다.

@Component
@Aspect
public class SampleAspect {

    @Pointcut("@annotation(com.edu.aop.Sample)")
    private void pointcut() { }

    @AfterReturning(pointcut = "pointcut()", returning = "returnValue")
    public Object afterReturning(JoinPoint joinPoint, Object returnValue) {
                //TODO : Something logic...
        return returnValue;
    }
}

 모든 어드바이스는 org.aspectj.lang.JoinPoint 타입의 파라미터를 어드바이스의 첫 파라미터로 선언할 수 있다. Around advice 는 JoinPoint 의 하위 클래스인 ProceedingJoinPoint 타입의 파라미터를 필수적으로 첫 파라미터로 선언해야 한다.

 JoinPoint 인터페이스는 getArgs() (메서드 argument를 반환한다.), getSignature() (advice되는 메서드의 시그니처를 반환한다.), toString() (advice 되는 메서드의 유용한 설명을 출력한다.) 같은 다수의 유용한 메서드를 제공한다.

지난 번에 이어서 Advice가 어떤 JoinPoint에 사용될 것인지를 지정하는 PointCut 표현식을 정리하겠습니다.

포인트컷에는 다양한 명시자를 이용할 수 있습니다.

참고 자료

https://groovysunday.tistory.com/201

https://snoopy81.tistory.com/296

https://sjh836.tistory.com/157

http://blog.naver.com/PostView.nhn?blogId=chocolleto&logNo=30086024618&categoryNo=29&viewDate=¤tPage=1&listtype=0

https://icarus8050.tistory.com/7?category=340924