티포의개발일지

이전 포스트와 다르게 Docker를 활용한 SSL 인증을 구현해보았다. Cerbot 컨테이너를 만들고, docker compose 명령어를 통해 동작

1. /docker/nginx/default.conf 생성

최초 certbot 인증 시에는 아래와 같이 기본 세팅으로 해야한다.

server {
    listen 80;
    listen [::]:80;

    server_name 도메인 이름;
    server_tokens off;

    location /.well-known/acme-challenge/ {
        root /var/www/certbot;
    }

}

2. /docker/nginx/ Dockerfile 생성

FROM nginx:latest
COPY ./default.conf /etc/nginx/conf.d/default.conf

CMD ["nginx", "-g", "daemon off;"]

3. docker-compose.yml 추가

services:
  nginx:
    build: ./docker/nginx
    ports:
      - "80:80"
      - "443:443"
    restart: always
    environment:
      TZ: Asia/Seoul
    volumes:
      - ${CERTBOT_WWW_DIR}:/var/www/certbot/:ro
      - ${CERTBOT_CONF_DIR}:/etc/nginx/ssl/:ro
    depends_on:
      - client
      - server

  certbot:
    image: certbot/certbot:latest
    volumes:
      - ${CERTBOT_WWW_DIR}:/var/www/certbot/:rw
      - ${CERTBOT_CONF_DIR}:/etc/letsencrypt/:rw
      
  client:
  	...
  server:
  	...

4. .env파일 작성

...

# CERTBOT
CERTBOT_WWW_DIR=/data/certbot/www
CERTBOT_CONF_DIR=/data/certbot/conf

5. Docker compose up

$ docker compose up -d

인증서 발급을 위해 Nginx 서버를 켜준다.

6. 인증서 발급

$ docker compose run --rm  certbot certonly --webroot --webroot-path /var/www/certbot/ -d 도메인이름

pem 키 위치 확인

7. default.conf 파일 수정

upstream client{
  server client:3000;
}

upstream server{
  server server:8080;
}

server {
    listen       80;
    listen  [::]:80;
    server_name  [도메인이름];

    location ^~ /.well-known/acme-challenge/ {
      default_type "text/plain";
      root /var/www/certbot;
    }

    location / {
        return 301 https://[도메인이름]$request_uri;
    }
}

server {

        listen 443 ssl default_server;
        listen [::]:443 ssl default_server;
        server_name  [도메인이름];

        ssl_certificate /etc/nginx/ssl/live/[도메인이름]/fullchain.pem;
        ssl_certificate_key /etc/nginx/ssl/live/[도메인이름]/privkey.pem;

        #access_log  /var/log/nginx/host.access.log  main;
        location /api {
            proxy_pass http://server;
            proxy_redirect off;
            rewrite ^/api/(.*)$ /$1 break;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection 'upgrade';
            proxy_set_header Host $host;
        }

        location / {
            proxy_pass http://client;
            root   /usr/share/nginx/html;
            index  index.html index.htm;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection 'upgrade';
            proxy_set_header Host $host;
        }

        #error_page  404              /404.html;

        # redirect server error pages to the static page /50x.html
        #
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   /usr/share/nginx/html;
        }

        #allow deny ip
        #allow 12.0.0.0;
        #deny all;
}

8. docker 재시작

$ docker compose up -d

Nginx Service에 restart:always 옵션이 존재하기 때문에 Nginx Service는 재시작된다.

9. 인증서 관련 명령어

- 인증서 확인

$ docker compose run --rm certbot certificates

- 인증서 갱신

$ docker compose run --rm certbot renew

10. 크론 탭 활용 (crontab)

- /bin/letsencrypt.sh 파일 작성

cd /home
date >> /home/certbot_renew.log
sudo docker compose run --rm certbot renew >> /home/certbot_renew.log
sudo docker compose restart nginx

- 실행 권한 부여

$ sudo chmod +x /bin/letsencrypt.sh

- 크론탭 열고 편집 

$ sudo crontab -e

- 아래 배치잡 생성

30 4 * * 0 /bin/letsencrypt.sh

- 저장 후 배치잡 확인

$ sudo crontab -l

-저장하고 크론 다시 실행

$ sudo service cron restart

1. rsync 다운로드

$ sudo yum install rsync

# 혹은

$ sudo apt install rsync

2. 사용법

$ rsync [OPTIONS] [SOURCE] [TARGET]

3. 예제

# 로컬 데이터를 로컬에 복사
$ rsync -avh /home/user/data /home/new_user/backup

# 로컬의 데이터를 리모트로 복사 
$ rsync -avh /home/user/data remote_user@remotehost:/home/remote_user/backup

# ssh 포트가 다른 경우
$ rsync -avh -e "ssh -p 123" /home/user/data remote_user@remotehost:/home/remote_user/backup

# 리모트 데이터를 로컬로 가져옴
$ rsync -avh remote_user@remotehost:/home/remote_user/backup /home/user/data

-ravPh 옵션을 주면 Progress도 확인할 수 있다.

4. ssh 파일 전송 예제

rsync -ravPh abc.tar Server:~/

1. ~/.ssh/config 생성

Host Server
  HostName IP주소
  User 유저이름
  Port 22
  IdentityFile pem파일위치

2. pem 파일 권한 수정

chmod 400 pem파일위치

3. 이름으로 ssh 접속

$ssh Server

Kafka SASL에 관한 정보는 구글링해도 잘 나온다.

하지만 세팅하는 부분은 생각보다 별로 없어서 여기 적어놔야겠다.

1. Kafka Broker Docker Compose

  kafka:
    image: bitnami/kafka:3.4
    hostname: kafka
    ports:
      - 9092:9092
    environment:
      KAFKA_CLIENT_USERS: user
      KAFKA_CLIENT_PASSWORDS: pass
      ALLOW_PLAINTEXT_LISTENER: yes
      KAFKA_BROKER_ID: 1
      KAFKA_CFG_PROCESS_ROLES: broker,controller
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: CLIENT
      KAFKA_CFG_LISTENERS: CONTROLLER://:9093,CLIENT://:9092
      KAFKA_CFG_ADVERTISED_LISTENERS: CLIENT://host.docker.internal:9092
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,CLIENT:SASL_PLAINTEXT
      KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CFG_DELETE_TOPIC_ENABLE: true
      KAFKA_CFG_BROKER_ID: 1
      KAFKA_CFG_NODE_ID: 1
      KRAFT_MODE: true
      KAFKA_ENABLE_KRAFT: yes
      KAFKA_MESSAGE_MAX_BYTES: 2000000000
      TZ: Asia/Seoul
    volumes:
      - ${KAFKA_DATA_DIR}:/bitnami/kafka/data

위와 같이 Users, Passwords를 추가해주고 ( 여러개여도 상관 없음. )

SASL에 관한 필드를 추가해준다.

2. Producer

  kafka:
    producer:
      bootstrap-servers: host.docker.internal:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="pass!";

3. Consumer

  kafka:
    consumer:
      bootstrap-servers: kafka:9092
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="pass!";

kafkaTemplate.send() 메소드에는 훅이 없다. 혹여나 Kafka 서버가 다운될 경우, kafkaTemplate.send() 메소드를 호출 자체를 막는게 좋은데, 마땅한 조건이 없을 경우 다음과 같은 함수를 만들어 검증해보자.

public boolean verifyConnection() throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrap);
        props.put("request.timeout.ms", 3000);
        props.put("connections.max.idle.ms", 5000);

        AdminClient client = AdminClient.create(props);
        Collection<Node> nodes = client.describeCluster()
            .nodes()
            .get();
        return nodes != null && nodes.size() > 0;
    }

또한, try catch절로 예외처리를 할 수도 있다.

    public boolean verifyConnection(String message, MessageType type) {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrap);
        props.put("request.timeout.ms", 500);
        props.put("connections.max.idle.ms", 900);

        AdminClient client = AdminClient.create(props);
        try {
            client.describeCluster()
                .nodes()
                .get();
        } catch (InterruptedException | ExecutionException e) {
            
            return false;
        }
        return true;
    }

이 방법이 마음에 안들면 정석은 아니지만 Sockek 객체로 해당 포트가 살아있는지 확인하는 방법도 있다.

Socket socket = new Socket();
        String[] bootstrapAddress = bootstrapServers.split(":");
        try {
            socket.connect(new InetSocketAddress(bootstrapAddress[0],
                Integer.parseInt(bootstrapAddress[1])));
            socket.close();
            return true;
        } catch (Exception e) {
            log.error("[Kafka Server is not available] message = {}", e.getMessage());
            return false;
        }

회사에서 API 게이트웨이 서버를 만들게 되었다.

검색을 해도 중구난방 잘 안되어있어서 경험하면서 기본적인 틀을 기록하고자 한다.

1. build.gradle에 추가해준다.

ext {
    set('snippetsDir', file("build/generated-snippets"))
    set('springCloudVersion', "2022.0.4")
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}

dependencies {

	...
    
    implementation 'org.springframework.boot:spring-boot-starter-data-r2dbc'
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.springframework.cloud:spring-cloud-starter-gateway'
    compileOnly 'org.projectlombok:lombok'
    runtimeOnly 'org.postgresql:postgresql'
    runtimeOnly 'org.postgresql:r2dbc-postgresql'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'io.projectreactor:reactor-test'
}

2. application-properties.yml 파일 수정

server:
  port: 8070

spring:
  jackson:
    timezone: Asia/Seoul
  data:
    r2dbc:
      repositories:
        enabled: true
  datasource:
    url: r2dbc:postgresql://localhost:5433/postgres
    username: postgres
    password: postgres
    driver-class-name: org.postgresql.Driver
  r2dbc:
    url: r2dbc:postgresql://localhost:5433/postgres
    username: postgres
    password: postgres
  cloud:
    gateway:
      default-filters: # Gateway 공통 필터
        - name: GlobalFilter
          args:
            baseMessage: hello world
      routes:
        - id: router-1
          uri: http://localhost:3000
          predicates:
            - Path=/**

비동기 서버를 위한 r2dbc를 사용해야 한다.

3. GlobalFilter 생성

@Slf4j
@Component
public class GlobalFilter extends AbstractGatewayFilterFactory<FilterDto> {

    public GlobalFilter() {
        super(FilterDto.class);
    }

    @Override
    public GatewayFilter apply(FilterDto dto) {
        return (exchange, chain) -> {
            log.info("GlobalFilter baseMessage: {}", dto.getMessage());
            return chain.filter(exchange).then(Mono.fromRunnable(() -> {
                log.info("GlobalFilter End: {}", exchange.getResponse());
            }));
        };
    }
}

4. FilterDto 생성

@Getter
public class FilterDto {

    private String message;
}

5. ApiRoute Entity 생성

@Entity
@Getter
@Table(name = "api_route")
@TableGenerator(name = "api_route", allocationSize = 1)
public class ApiRoute {

    @Id
    @GeneratedValue(strategy = GenerationType.AUTO, generator = "api_route_seq")
    @SequenceGenerator(name = "api_route_seq", sequenceName = "api_route_seq", allocationSize = 1)

    @Column(name = "id")
    private String id;

    @Column
    private String routeIdentifier;
    @Column
    private String uri;
    @Column
    private String method;
    @Column
    private String path;
}

6. ApiRouteRepository 생성

@Repository
public interface ApiRouteRepository extends R2dbcRepository<ApiRoute, String> {

}

7. ApiRouteService 생성

Service Interface는 아래의 Override 한 메소드만 추가해주면 된다.

@Service
@RequiredArgsConstructor
public class ApiRouteServiceImpl implements ApiRouteService {

    private final ApiRouteRepository apiRouteRepository;

    @Override
    public Flux<ApiRoute> getAll() {
        return this.apiRouteRepository.findAll();
    }

    public Mono<ApiRoute> create(ApiRoute apiRoute) {
        return this.apiRouteRepository.save(apiRoute);
    }

    public Mono<ApiRoute> getById(String id) {
        return this.apiRouteRepository.findById(id);
    }
}

8. ApiPathRouteLocatorImpl 생성

@AllArgsConstructor
public class ApiPathRouteLocatorImpl implements RouteLocator {

    private final ApiRouteService apiRouteService;
    private final RouteLocatorBuilder routeLocatorBuilder;

    @Override
    public Flux<Route> getRoutes() {
        RouteLocatorBuilder.Builder routesBuilder = routeLocatorBuilder.routes();
        return apiRouteService.getAll()
            .map(apiRoute -> routesBuilder.route(String.valueOf(apiRoute.getRouteIdentifier()),
                predicateSpec -> setPredicateSpec(apiRoute, predicateSpec)))
            .collectList()
            .flatMapMany(builders -> routesBuilder.build()
                .getRoutes());
    }

    private Buildable<Route> setPredicateSpec(ApiRoute apiRoute, PredicateSpec predicateSpec) {
        BooleanSpec booleanSpec = predicateSpec.path(apiRoute.getPath());
        if (!StringUtils.isEmpty(apiRoute.getMethod())) {
            booleanSpec.and()
                .method(apiRoute.getMethod());
        }
        return booleanSpec.uri(apiRoute.getUri());
    }

    @Override
    public Flux<Route> getRoutesByMetadata(Map<String, Object> metadata) {
        return RouteLocator.super.getRoutesByMetadata(metadata);
    }
}

9. GatewayConfig 생성

@Configuration
@Slf4j
public class GatewayConfig {

    @Bean
    public RouteLocator routeLocator(ApiRouteService routeService,
        RouteLocatorBuilder routeLocationBuilder) {
        return new ApiPathRouteLocatorImpl(routeService, routeLocationBuilder);
    }

}

여기까지가 기본적인 프록시를 위한 라우터이다.

다음 부터는 API로 route를 CRUD 하기 위한 작업이다.

10. ApiRouteRouter Configuration 생성

@Configuration
public class ApiRouteRouter {

    @Bean
    public RouterFunction<ServerResponse> route(ApiRouteHandler apiRouteHandler) {
        return RouterFunctions.route(POST("/routes")
                .and(accept(MediaType.APPLICATION_JSON)), apiRouteHandler::create)
            .andRoute(GET("/routes/{routeId}")
                .and(accept(MediaType.APPLICATION_JSON)), apiRouteHandler::getById)
            .andRoute(GET("/routes/refresh-routes")
                .and(accept(MediaType.APPLICATION_JSON)), apiRouteHandler::refreshRoutes);
    }
}

11. ApiROuteHandler 생성

@RequiredArgsConstructor
@Component
@Slf4j
public class ApiRouteHandler {

    private final ApiRouteService routeService;

    private final GatewayRoutesRefresher gatewayRoutesRefresher;

    public Mono<ServerResponse> create(ServerRequest serverRequest) {
        Mono<ApiRoute> apiRoute = serverRequest.bodyToMono(ApiRoute.class);
        return apiRoute.flatMap(route ->
            ServerResponse.status(HttpStatus.OK)
                .contentType(MediaType.APPLICATION_JSON)
                .body(routeService.create(route), ApiRoute.class));
    }

    public Mono<ServerResponse> getById(ServerRequest serverRequest) {
        log.info("serverRequest.pathVariable(\"routeId\") = {}",
            serverRequest.pathVariable("routeId"));
        final String apiId = serverRequest.pathVariable("routeId");
        Mono<ApiRoute> apiRoute = routeService.getById(apiId);
        return apiRoute.flatMap(route -> ServerResponse.ok()
                .body(fromValue(route)))
            .switchIfEmpty(ServerResponse.notFound()
                .build());
    }

    public Mono<ServerResponse> refreshRoutes(ServerRequest serverRequest) {
        gatewayRoutesRefresher.refreshRoutes();
        return ServerResponse.ok().body(BodyInserters.fromObject("Routes reloaded successfully"));
    }
}

12. GatewayRoutesRefresher 생성

@Component
public class GatewayRoutesRefresher implements ApplicationEventPublisherAware {

    private ApplicationEventPublisher applicationEventPublisher;

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    /**
     * Refresh the routes to load from data store
     */
    public void refreshRoutes() {
        applicationEventPublisher.publishEvent(new RefreshRoutesEvent(this));
    }
}

만약, domain Hostname에 따라 Proxy되는 서버의 주소를 변경하고자 하면 다음과 같은 방법을 쓸 수 있다.

@Override
    public GatewayFilter apply(HostNameFilterDto hostNameFilterdto) {

        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();

            request.getHeaders();

            String[] split = request.getURI().getHost().split("\\.");
            String siteId = split[0];

            Mono<Site> siteMono = siteService.findById(siteId);
            Site site = siteMono.share().block();

            String uri = Objects.requireNonNull(site).getConnectHost();
            int port = site.getConnectPort();
            if (port != 80) {
                uri += ":" + port;
            }

            Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
            Route newRoute = Route.async()
                .id(site.getId())
                .uri(uri)
                .predicate(serverWebExchange -> false)
                .order(Objects.requireNonNull(route).getOrder())
                .filters(route.getFilters())
                .build();
            exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, newRoute);

            return chain.filter(exchange);
        };
    }

Dockerfile로 필요한 confluent 패키지를 다운로드 받고, add_connector.sh 와 connect-distributed.properties 파일을 사용한다.

- add_connector.sh 에선 Kafka 연결을 확인하고 connect-mqtt-source.json파일대로 curl을 실행한다.

- connect-distributed.properties 파일로 connect 기본 설정을 한다.

1. Kafka, Kafka-Connect Docker Compose 생성

  kafka:
    image: bitnami/kafka:3.4
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      ALLOW_PLAINTEXT_LISTENER: yes
      KAFKA_BROKER_ID: 1
      KAFKA_CFG_PROCESS_ROLES: broker,controller
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: CLIENT
      KAFKA_CFG_LISTENERS: CONTROLLER://:9093,CLIENT://:9092
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,CLIENT:PLAINTEXT
      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: CLIENT://kafka:9092
      KAFKA_CFG_DELETE_TOPIC_ENABLE: true
      KAFKA_CFG_BROKER_ID: 1
      KAFKA_CFG_NODE_ID: 1
      KRAFT_MODE: true
      KAFKA_ENABLE_KRAFT: yes
      TZ: Asia/Seoul
    volumes:
      - ${KAFKA_DATA_DIR}:/bitnami/kafka/data

  kafka-connect:
    build: ./docker/kafka-connect
    container_name: kafka-connect
    ports:
      - "8083:8083"
    depends_on:
      - kafka
    volumes:
      - ${KAFKA_CONNECT_DATA_DIR}:/var/lib/kafka-connect

2. Kafka Connect Dockerfile 생성

프로젝트/docker/kafka-connect

FROM bitnami/kafka:3.4

ARG MODULE=kafka-connect
ENV MODULE=${MODULE}

USER root

WORKDIR /connectors

RUN apt-get update && apt-get upgrade -y && apt-get install unzip -y

RUN curl -O https://client.hub.confluent.io/confluent-hub-client-latest.tar.gz
RUN tar -zxvf confluent-hub-client-latest.tar.gz
RUN rm -f confluent-hub-client-latest.tar.gz
RUN mkdir component && mkdir config && cd config && touch worker.properties
RUN cd /connectors/bin && ./confluent-hub install --no-prompt confluentinc/kafka-connect-mqtt:1.7.1 \
    --component-dir /connectors/component \
    --worker-configs /connectors/config/worker.properties
RUN cp -p /connectors/component/confluentinc-kafka-connect-mqtt/lib/* /opt/bitnami/kafka/libs

WORKDIR /tmp
COPY . .
CMD /tmp/add_connector.sh & /opt/bitnami/kafka/bin/connect-distributed.sh /tmp/connect-distributed.properties

3. connect-distributed.properties 생성

bootstrap.servers=kafka:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
converter.encoding=UTF-8
offset.flush.interval.ms=10000
topic.prefix=test
tasks.max=5
plugin.path=/connectors

4. connect-mqtt-source.json 생성

{
  "name": "mqtt-source-1",
  "config": {
    "bootstrap.servers": "kafka:9092",
    "connector.class": "io.confluent.connect.mqtt.MqttSourceConnector",
    "tasks.max": "1",
    "mqtt.server.uri": "tcp://host.docker.internal:9000",
    "mqtt.topics": "data",
    "kafka.topic": "mqtt.data",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "confluent.topic.bootstrap.servers": "kafka:9092",
    "confluent.topic.replication.factor": 1
  }
}

5. add_connector.sh 생성

# Wait for Kafka Connect to start
while true
do
  status=$(curl -s -o /dev/null -w %{http_code} http://localhost:8083/connectors)
  if [ "$status" -eq 200 ]; then
    echo "Kafka Connect has started."
    break
  else
    echo "Waiting for Kafka Connect to start..."
    sleep 3
  fi
done


echo "Start to add kafka-connect"

# ADD Mqtt Source Connectors
curl -d @/tmp/connect-mqtt-source.json -H "Content-Type: application/json" -X POST http://localhost:8083/connectors

echo "Complete to add kafka-connect"

6. Kafka consumer (Spring Boot) 사용법은 아래를 참고한다.

Spring boot | Spring Apache Kafka 사용법 ( with Docker Container ) | Consumer :: 티포의개발일지 (tistory.com)

7. docker compose 실행 후 mqtt에서 보낸 데이터를 확인한다.

1. Mosquitto Docker compose 설정

  mosquitto:
    container_name: lpms-mosquitto
    restart: always
    image: eclipse-mosquitto
    ports:
      - "9000:1883"
      - "9001:9001"
    volumes:
      - ${MOSQUITTO_DIR}/config/mosquitto.conf:/mosquitto/config/mosquitto.conf
      - ${MOSQUITTO_DIR}/data:/mosquitto/data
      - ${MOSQUITTO_DIR}/log:/mosquitto/log

- mosquitto.conf

allow_anonymous true
connection_messages true
log_type all
listener 1883

2. Spring boot build.gradle 추가

    implementation 'org.springframework.boot:spring-boot-starter-integration'
    implementation 'org.springframework.integration:spring-integration-mqtt'

3. MqttProperties 클래스 생성

@ConfigurationProperties(prefix = "mqtt")
@Component
@Data
@Validated
public class MqttProperties {

    private String name;
    private String password;
    private String url;
    private Integer qos;
    private String topic;

}

4. MqttConfig 클래스 생성

본 프로젝트에서는 outBound만 사용 할 예정이다.

@Configuration
@RequiredArgsConstructor
public class MqttConfig {

    private static final String MQTT_CLIENT_ID = MqttAsyncClient.generateClientId();
    private final MqttProperties properties;

    /**
     * DefaultMqttPahoClientFactory를 통해 MQTT 클라이언트를 등록
     */
    @Bean
    public DefaultMqttPahoClientFactory defaultMqttPahoClientFactory() {
        DefaultMqttPahoClientFactory clientFactory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setServerURIs(new String[]{properties.getUrl()});
        options.setUserName(properties.getName());
        options.setPassword(properties.getPassword().toCharArray());
        clientFactory.setConnectionOptions(options);
        return clientFactory;
    }

    /**
     * MQTT 클라이언트를 통해 메시지를 구독하기 위하여 MqttPahoMessageDrivenChannelAdapter를 통해 메시지 수신을 위한 채널을 구성
     */
//    @Bean
//    public MessageChannel mqttInputChannel() {
//        return new DirectChannel();
//    }
//
//    @Bean
//    public MessageProducer inboundChannel() {
//        MqttPahoMessageDrivenChannelAdapter adapter =
//            new MqttPahoMessageDrivenChannelAdapter(
//                properties.getUrl(),
//                MQTT_CLIENT_ID,
//                properties.getTopic());
//        adapter.setCompletionTimeout(5000);
//        adapter.setConverter(new DefaultPahoMessageConverter());
//        adapter.setQos(1);
//        adapter.setOutputChannel(mqttInputChannel());
//        return adapter;
//    }
//
//    @Bean
//    @ServiceActivator(inputChannel = "mqttInputChannel")
//    public MessageHandler inboundMessageHandler() {
//        return message -> {
//            String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
//            System.out.println("Topic:" + topic);
//            System.out.println("Payload" + message.getPayload());
//        };
//    }

    /**
     * Message outbound를 위한 채널 구성
     */

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound(DefaultMqttPahoClientFactory clientFactory) {
        MqttPahoMessageHandler messageHandler =
            new MqttPahoMessageHandler(MQTT_CLIENT_ID, clientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(1);
        return messageHandler;
    }

    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MyGateway {

        void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);

    }

}

5. MqttService 클래스 생성

@Service
@RequiredArgsConstructor
public class MqttService {

    private final MyGateway myGateway;

    public void send() {
        myGateway.sendToMqtt("12345", "/a/b/q");
    }
}

6. 도커 컨테이너 만들고, Spring 서버 킨 다음 필요한 곳에서 사용

1. 프로세스는 다음과 같다.

1. Git Repository에 push 또는 merge
2. Github Actions에서 이를 감지하고 빌드 및 테스트 실행
3. 테스트까지 실행될 경우 Docker 이미지로 빌드
4. 빌드된 이미지를 Dockerhub에 업로드
5. EC2 Instance에 ssh로 접속 후 이미지를 pull

2. 파이프라인 구축

1. Github에 Repository를 만들고 프로젝트를 업로드한다.

2. Dockerfile을 만들어준다.

3. Github에 들어가 Actions에서 Java with Gradle configure버튼을 클릭한다.

4. 원하는 yml 파일 이름을 작성하고 아래 코드를 붙여넣는다.

# Workflow 이름
name: Spring Boot & Gradle CI/CD

# 어떤 이벤트가 발생하면 workflow 실행할 지 명시
on:
  # main 브랜치에 push나 pull request 발생 시
  push:
    branches: [ "main" ]
  pull_request:
    branches: [ "main" ]
    
permissions:
  contents: write
  
# 위 이벤트 발생 시 실행될 작업들
jobs:
  build:
    # VM의실행 환경 지정 => 우분투 최신 버전
    runs-on: ubuntu-latest
    
    # working directory
    defaults:
    	run:
        	working-directory: ./backend

    # 실행될 jobs를 순서대로 명시
    steps:
    - name: Checkout
      uses: actions/checkout@v3

    # JDK 11 설치
    - name: Set up JDK 11
      uses: actions/setup-java@v3
      with:
        java-version: '11'
        distribution: 'temurin'

    # Gradle Build를 위한 권한 부여
    - name: Grant execute permission for gradlew
      run: chmod +x gradlew

    # Gradle Build (test 제외)
    - name: Build with Gradle
      run: ./gradlew clean build --exclude-task test
      
    # 테스트 결과를 코멘트로 달아줌
    - name: Publish test result
      uses: EnricoMi/publish-unit-test-result-action@v1
      if: always()
      with:
        files: 'build/test-results/test/*.xml'
        
	# 테스트 실패 시 해당 부분에 코멘트 달아줌.
    - name: When test fail, a comment is registered in the error code line.
      uses: mikepenz/action-junit-report@v3
      if: always()
      with:
        report_paths: 'build/test-results/test/*.xml'

    # DockerHub 로그인
    - name: DockerHub Login
      uses: docker/login-action@v2
      with:
        username: ${{ secrets.DOCKERHUB_USERNAME }}
        password: ${{ secrets.DOCKERHUB_PASSWORD }}

    # Docker 이미지 빌드
    - name: Docker Image Build
      run: docker build -t ${{ secrets.DOCKERHUB_USERNAME }}/${{ secrets.PROJECT_NAME }} .

    # DockerHub Push
    - name: DockerHub Push
      run: docker push ${{ secrets.DOCKERHUB_USERNAME }}/${{ secrets.PROJECT_NAME }}

    # EC2 인스턴스 접속 및 애플리케이션 실행
    - name: Application Run
      uses: appleboy/ssh-action@v0.1.6
      with:
        host: ${{ secrets.EC2_HOST }}
        username: ${{ secrets.EC2_USERNAME }}
        key: ${{ secrets.EC2_KEY }}

        script: |
          sudo docker kill ${{ secrets.PROJECT_NAME }}
          sudo docker rm -f ${{ secrets.PROJECT_NAME }}
          sudo docker rmi ${{ secrets.DOCKERHUB_USERNAME }}/${{ secrets.PROJECT_NAME }}
          sudo docker pull ${{ secrets.DOCKERHUB_USERNAME }}/${{ secrets.PROJECT_NAME }}

          sudo docker run -p ${{ secrets.PORT }}:${{ secrets.PORT }} \
          --name ${{ secrets.PROJECT_NAME }} \
          -e SPRING_DATASOURCE_URL=${{ secrets.DB_URL }} \
          -e SPRING_DATASOURCE_USERNAME=${{ secrets.DB_USERNAME }} \
          -e SPRING_DATASOURCE_PASSWORD=${{ secrets.DB_PASSWORD }} \
          -d ${{ secrets.DOCKERHUB_USERNAME }}/${{ secrets.PROJECT_NAME }}

• ${{}} 변수는 환경변수이다.
• 완성되면 .github/workflows 경로에 yml 파일이 추가된다.
• 수정하고 싶으면 이 파일을 수정하고, 다른 workflow 파일을 작성해도 된다.
• 만약 안된다면 Github Settings -> Developer Settings -> Personal access tokens에서 생성한 토큰에 들어가 workflow를 체크해주면 된다.

5. Settings -> Secrets and variables -> Actions 에서 환경변수 추가

• DOCKERHUB_USERNAME : 본인의 Docker Hub Username
• DOCKERHUB_PASSWORD : 본인의 Docker Hub Password
• PROJECT_NAME : 프로젝트 이름 (ci-cd-practice) => 이 이름으로 Docker Hub에 올라가게 되고, Docker Container 이름도 이 이름으로 설정할 예정
• PORT : Docker을 실행시킬 포트 번호 (ex: 8080)
• DB_URL : 프로젝트에 사용될 DB의 URL (ex:  jdbc:mysql://RDS주소:DB포트/DB명)
• DB_USERNAME : 프로젝트에 사용될 DB의 Username (ex: root)
• DB_PASSWORD : 프로젝트에 사용될 DB의 Password
• EC2_HOST : AWS EC2 인스턴스의 퍼블릭 IPv4 DNS (ex: ec2-52-79-213-143.ap-northeast-2.compute.amazonaws.com)
• EC2_USERNAME : AWS EC2 인스턴스의 Username (ex: ubuntu)
• EC2_KEY : AWS EC2 인스턴스를 생성할 때 저장된 pem 키
  • MAC을 기준으로 터미널에서 cat <pem 키 경로>를 입력하면 (드래그앤 드롭해도 됨) '-----BEGIN RSA PRIVATE KEY-----'부터 '-----END RSA PRIVATE KEY-----'까지(맨 뒤에 %빼고)를 복사해서 이 값으로 넣어주면 됨

6. workflow를 실행해준다.

3. 테스트 실패 시 Merge 막기

Repository Settings → Branches → Add rule 을 선택한다.

Branch name pattern을 설정하고,
Require status checks to pass before merging 설정을 통해 merge를 위해 통과해야할 Action들을 선택할 수 있다.

설정 후에 다음과 같이 merge를 못하도록 막혀져있는 것을 확인할 수 있다.
(지금은 admin 이라 강제로 merge할 수 있게 merge 버튼이 활성화돼있지만, admin이 아닌 경우 merge 버튼이 비활성화 된다)

참고

[CI/CD] Github Actions를 활용한 CI/CD 파이프라인 구축 (+ Docker hub) (tistory.com)

Spring Security를 사용할 때 UserDetail 객체를 사용하는데, 예를들어

위와 같은 애들을 Request로 받아올 수 있다.

아래와 같이 사용하면 된다.