'분류 전체보기'에 해당되는 글 311건

Linux 환경의 경우, fluentd native나 td-agent 모두 stable 버전으로 제공되는데,

Windows 환경의 경우는 공식홈에서 밀고 있는 td-agent 조차 베타버전으로 제공이 되고 있는 상황이고,

Document를 샅샅히 뒤져보다보면, nxlog 라는 놈을 하나 소개해준다. (http://docs.fluentd.org/v0.12/articles/windows)

실제로 테스트를 해봐도 tg-agent랑 nxlog 둘다 써봤지만, 제대로되는 게 nxlog 뿐인것 같다.

하지만, nxlog에서는 fluentd의 forward protocol을 지원하지 않아, tcp를 이용해서 전송을 해야 했다.

1. 서버 설치 (docker)

### docker-compose.yml
version: '2'
services:
  fluentd-recv:
    container_name: 'fluentd-recv'
    image: fluent/fluentd
    volumes:
      - ./conf:/fluentd/etc
    ports:
      - 5170:5170

### conf/fluent.conf
# Input
<source>
  @type tcp     # tcp 프로토콜로 데이터를 받는다.
  port 5170      
  tag testlog    # testlog라는 식별자를 추가한다.
  format json   # json 포맷만 받음.
</source>

# Output
<match testlog> # testlog tag를 가진 데이터만 처리함
  @type file    # 파일로 저장
  path /fluentd/recv/* 
</match>

2. nxlog (Client) 설치

link : http://nxlog.co/products/nxlog-community-edition/download

위 링크에서 다운받아서, 설치하고 나면 C:\Program Files(x86)\nxlog 경로에 있는 nxlog.conf를 아래와 같이 수정한다.

### nxlog.conf
define ROOT C:\Program Files (x86)\nxlog

Moduledir %ROOT%\modules
CacheDir %ROOT%\data
Pidfile %ROOT%\data\nxlog.pid
SpoolDir %ROOT%\data
LogFile %ROOT%\data\nxlog.log

<Input in>
    Module      im_file      # fluentd의 input tail plugin의 역할을 함
	File		'D:\fluentd_test\send\\*.json'     # 해당 경로에 있는 파일들을 감시한다. (와일드카드를 사용하려면 \를 하나 더 붙여줘야 된다.)
	SavePos 	TRUE    # 파일의 position 관리 enable / disable
	InputType	LineBased	 # 파일 내용의 라인 단위로 파싱한다.
</Input>

<Output out>
    Module      om_tcp     # in으로 들어온 파일을 tcp형태로 전송한다.
    Host        192.168.0.10 # 전송할 서버 정보 입력
    Port        5170
</Output>

<route 1> # 위에 설정한 내용들을 어떻게 동작시킬지 지정한다.
    Path        in => out      # in으로 받아서 out으로 보낸다.
</route>

### 전송이 됬는지 안됬는지 판단이 어려우므로 테스트 시 추가한 설정 (선택사항) ###
# 파일을 tcp로 보내고 난 뒤, 내용을 지정한 경로에 파일로 저장한다. 
<Output outfile>
  Module om_file
  File   'D:\fluentd_test\send\backup.log'
</Output>

<route 2> # Route 1과 마찬가지로 동작방식 지정
    Path        in => outfile
</route>

n개의 클라이언트에서 발생하는 json로그를 서버 한 곳에서 보아야 할 일이 생겼는데,

해당 서버가 fluentd 방식을 지원한다고 하여, json 로그를 특정 폴더에 남겨서, 실시간으로 tailing 하여 fluentd의 forward 프로토콜을 이용하여 전송하는 방식으로 설정을 해보았다.

1. 서버 구성 (docker)

### docker-compose.yml
version: '2'
services:
  fluentd-recv:
    container_name: 'fluentd-recv'
    image: fluent/fluentd
    volumes:
      - ./conf:/fluentd/etc
    ports:
      - 24224:24224

### conf/fluent.conf
# Input 
<source>
  @type forward     # forward 프로토콜을 이용하여 전송받는다. (forward는 전송은 tcp로 하고, health check는 udp로 하는 방식)
  port 24224           # 24224 포트를 이용한한다.
</source>

# Output
<match **>     # 보내는 부분에서 tag를 지정하여, tag별로 설정 가능하다.
  @type file      # 받은 내용을 파일로 저장함
  path /fluentd/recv/*   # 이 경로에 저장함. (파일명을 명시할 수도 있는 것 같다.)
</match>

2. 클라 구성 (docker)

### docker-compose.yml
version: '2'
services:
  fluentd-send:
    container_name: 'fluentd-send'
    image: fluent/fluentd
    volumes:
      - ./conf:/fluentd/etc

### conf/fluent.conf
# Input
<source>
  @type tail     # 지정된 경로의 지정된 패턴에 해당하는 파일의 내용을 감시한다.
  path /fluentd/send/*.json     # 경로 지정 
  pos_file /fluentd/send/test_log.pos    # 파일의 position 관리 (이 설정이 없으면, 네트워크 이슈 발생 시 파일이 중복 전송되거나 잘려서 전송될 수 있으므로, 파일 내용의 정합성이 깨질 수 있다.)
  tag testlog    # 식별자 지정 (match에서 활용됨)
  format json  # 파일포맷 지정
  refresh_interval 5s      # 5초 주기로 감시한다.
</source>

# Output
<match testlog>   # tag가 testlog인 것만 아래 구문을 처리
  @type forward    # 전송 시 forward 프로토콜을 이용한다.
  flush_interval 5s # 5초 주기로 전송한다.
  <server>       # 전송할 서버 정보 입력
    host 192.168.0.10
    port 24224
  </server>
</match>

Ubuntu 환경에서 Package 설치를 기준으로 구성해볼 예정이다. (버전은 3.10)

1. 패키지 등록 및 설치

# apt-get source list 및 key 등록
$ echo "deb http://www.apache.org/dist/cassandra/debian 310x main" | sudo tee -a /etc/apt/sources.list.d/cassandra.sources.list
$ curl https://www.apache.org/dist/cassandra/KEYS | sudo apt-key add -
$ apt-get update

# 설치
$ apt-get install -y cassandra

2. 설정

제대로된 운영을 하려면 시스템에 맞게 건드려줘야 되는 항목들이 많기 때문에 클러스터 구성에 대한 최소한의 설정만 설명할 예정이다.

기본적으로 Consist-Hashing 방식으로 노드를 분산하며, gossip protocol을 이용하여 서로 노드간 alive 체크를 하기 때문에,

별도의 마스터가 없다는게 카산드라 클러스터 구성의 컨셉이다.

### /etc/cassandra.yaml
cluster_name : test_cluster         # 클러스터명 (클러스터가 여러개로 구성된 경우에 식별자로 사용되는 듯)
seed_provider:
  - class_name: org.apache.cassandra.locator.SimpleSeedProvider
    parameters:
      - seeds: 192.168.0.10  # 최초 초기화 시 leader 역할을 할 노드의 ip를 적어준다. 
                                      # datacenter간에 cluster구성을 해야 하는 경우
                                      # HA효과를 보기위해 datacenter별로 최소 한대씩은 적어주는 것을 권장한다. (콤마 구분)
listen_address: 192.168.0.10     # 현재 노드의 ip 적어준다. (docker구성인 경우 docker의 가상 ip 입력)
broadcast_address: 192.168.0.10 # 현재 노드의 ip 적어준다. (docker구성인 경우 docker가 동작하는 host 머신의 ip 입력)
rpc_address: 0.0.0.0  # rpc 통신을 위해 0.0.0.0을 입력해준다.
broadcast_rpc_address: 192.168.0.10 # broadcast_address와 동일
endpoint_snitch: GossipingPropertyFileSnitch # 클러스터 간 근접도 판단 방법에 대한 옵션으로 
                                                                # 자세한건 http://docs.datastax.com/en/archived/cassandra/2.0/cassandra/architecture/architectureSnitchesAbout_c.html를 참조한다.                                                           

3. 실행

$ service cassandra start

Kafka의 경우 Zookeeper가 클러스터의 역할을 대부분 수행(offset관리, failover)해주기 때문에,

Zookeeper 구성만 잘 되어 있다면 설정 레벨에서는 별로 신경쓸 것이 없다.

1. 설치

# download (다른 파일에 비해 압축파일 명이 좀 긴데 앞에 2.12가 scala버전이고 0.10.2가 kafka버전임)
$ wget http://apache.mirror.cdnetworks.com/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz

# 압축 해제
$ tar -zxvf kafka_2.12-0.10.2.1.tgz

# 심볼릭 링크 설정 (선택)
$ ln -s kafka_2.12-0.10.2.1 kafka

2. 설정

config 폴더에 보면 정말 많은 properties 파일이 있는데, kafka 서버 구동을 위해 사용되는 설정은 server.properties라고 보면된다.

# server.properties
broker.id=0            # 클러스터 구성 시 각 노드에 id값을 주어, 각 노드를 식별하는 역할 (다른 노드와 중복되면 안됨)
listeners=PLAINTEXT://192.168.0.1:9092      # 현재 호스트 머신의 IP를 적어준다. 앞에 PLAINTEXT 부분은 kafka에서 지원하는 전송 프로토콜로, 자세한건 공식홈을 참조하는 것이 정확하다.
advertised.listeners=PLAINTEXT://192.168.0.1:9092   # 현재 호스트 머신의 IP를 적어준다. (Producer, Consumer가 참조하게 되는 IP)
...
zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181    # kafka를 관리해줄 zookeeper목록을 적는다. (콤마 구분)
...

위에 설명한 정도만 수정했고, 나머지는 default로 두어도 무방하다. (대규모 서비스는 micro control이 필요할지도..)

3. 실행

bin 폴더에 친절하게도 여러가지 스크립트들이 제공되어 있다. (백그라운드 동작은 아니어서 nohup으로 띄워야 한다.)

만약 zookeeper를 별도로 세팅하지 않았더라도, 자체 제공하는 zookeeper가 있어서 사용은 가능하지만, 권장하진 않고 cluster 구성도 제대로 되지는 않는 것 같다.

단, single node로 kafka를 운영하고자 할 경우에는 embedded zookeeper도 나쁘진 않다.

$ nohup bin/kafka-server-start.sh config/server.properties &

다른 노드의 경우도 broker.id만 겹치지 않게 여러개 띄워주면 되고, client 쪽에서만 kafka 서버들의 목록을 잘 넣어주면 된다.

# download
wget http://apache.mirror.cdnetworks.com/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz

# 압축해제
tar -zxvf zookeeper-3.4.10.tar.gz

# 심볼릭 링크 설정 (선택)
ln -s zookeeper-3.4.10 zookeeper

설치하고 나면, conf 폴더에 zoo_sample.cfg 파일이 있는데, 복사하여 zoo.cfg파일을 만든 후 내용을 아래와 같이 편집한다.

# zoo.cfg
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# for cluster
server.1=0.0.0.0:2888:3888
server.2=머신2의 ip:2888:3888
server.3=머신3의 ip:2888:3888

for cluster 주석 아래 부분에 server목록을 지정해주는데, server.{id} 형태로 지정하게 된다.

id 부분에 입력된 숫자를 dataDir에 myid 파일을 생성하여 아래와 같이 입력해주어야 한다.

$ echo 1 >> /tmp/zookeeper/myid

server.1의 경우 ip가 0.0.0.0인데, 해당 하는 노드가 localhost에 위치해 있는 경우, 예외상황 발생을 막기 위해 0.0.0.0으로 지정해주는 것을 권장한다고 하니, 자기 자신은 0.0.0.0으로 지정해준다.

이제 Zookeeper 서버를 띄워본다.

$ bin/zkServer.sh start

서버를 띄운 후 설정 값에 문제가 있는 경우, Cannot open channel to x at election address 와 같은 에러를 로그에서 확인할 수 있는데, 이런 경우는 myid 값이 제대로 들어가지 않았거나, 존재하지 않는 경우 발생하니 다시 한번 체크해보면 된다.

application/json 헤더를 지정하여 WebAPI를 통해 들어오는 Json Request Body를 그대로 Broker에 전달해야하는 API를 만들 일이 생겼다.

전달하는 것까지는 좋은데, 최소한의 검증 (Null, Empty Check나 integer의 경우 마이너스 값 체크) 정도는 해야되겠다 싶어서, @Valid를 이용해보기로 하였다.

일단 Model선언은 아래와 같이 했다.

public class TestLog {
	@Min(1)
	private int id;
	
	@NotNull
	@NotEmpty	
	private String description;
	
	@NotNull
	@NotEmpty
	private String now;
}

@Min, @NotNull 의 경우 기본적으로 java에서 제공하지만, @NotEmpty같은 경우는 hibernate-validator 의존성을 추가해주어야 한다.  (http://hibernate.org/validator/)

위와 같이 Model을 만든 후 Controller의 API에, @Valid만 붙여주면 된다.

@PostMapping("/test_log")
public Result testLog(@RequestBody @Valid TestLog testLog) {
	return "ok";
}

이제 Body로 들어오는 파라미터가 id는 1보다 작은 경우, description, now는 비어있거나 Null인 경우 400에러와 함께, org.springframework.web.bind.MethodArgumentNotValidException를 뱉어내게 된다.

클라 입장에서 받는 Response Body는 아래와 같이 생겼다.

{
	"timestamp": 1494829780576,
	"status": 400,
	"error": "Bad Request",
	"exception": "org.springframework.web.bind.MethodArgumentNotValidException",
	"errors": [{
		"codes": ["Min.testLog.id", "Min.id", "Min.int", "Min"],
		"arguments": [{
			"codes": ["testLog.id", "id"],
			"arguments": null,
			"defaultMessage": "id",
			"code": "id"
		}, 1],
		"defaultMessage": "반드시 1보다 같거나 커야 합니다.",
		"objectName": "testLog",
		"field": "id",
		"rejectedValue": 0,
		"bindingFailure": false,
		"code": "Min"
	}],
	"message": "Validation failed for object='testLog'. Error count: 1",
	"path": "/api/test_log"
}

기본적으로 위와 같은 Json 형태로 Response를 주기 때문에, 클라 입장에서 공통으로 Body를 parsing하여 사용하려면, 서버 단에서 해당 Exception을 핸들링하여 별도로 아래와 같이 예외처리를 해주어야 한다. 

@ExceptionHandler(MethodArgumentNotValidException.class)
@ResponseStatus(HttpStatus.BAD_REQUEST)
public Result handleMethodArgumentException(MethodArgumentNotValidException e) {	
        return "fail";
}

apt-get이나 yum을 이용해서 패키지 설치를 하게 되면, 기본적으로 cassandra는 127.0.0.1로 뜨게되어,

로컬 환경에서밖에 접속을 못하게 된다.

이런 경우, cassandra.yaml을 열어서 rpc가 지원되도록 아래와 같은 설정을 찾아서 변경해주어야 한다.

콜론 뒤에 띄어쓰기가 무조건 한 칸 있어야 에러가 발생하지 않으므로, 설정 변경 간 띄어쓰기에 유의해야 한다.

# /etc/cassandra/cassandra.yaml
start_rpc: true     # default : false
rpc_address: 0.0.0.0  # default: localhost
broadcast_rpc_address: localhost  # 원래 주석처리 되어있던 걸 풀어서 localhost로 맞춰준다.

이렇게 설정해두면, 기본 포트인 9042포트와 Thrift를 이용한 고속 전송(맞는 표현인지 모르겠음;)을 지원하는 9160포트가 동시에 0.0.0.0으로 listening 되게 된다.

cassandra.yaml에서만 변경하면, 서비스 재시작 시 자동으로 반영해줄거라 생각했지만, 클러스터 명이 다르다는 에러가 발생하게 된다.

이런 경우 설정을 원래대로 원복시키고, cqlsh에 들어가서 직접 쿼리를 날려준 후, 설정 파일을 바꾸어주면 문제가 발생하지 않는다.

cqlsh> UPDATE system.local SET cluster_name = '변경할 클러스터명' where key='local';

docker-compose를 사용했다는 가정하에, Version2의 compose파일을 이용하는 경우는 아래와 같이 확인 (브릿지어댑터를 이용해서 네트워크 할당을 한다고 한다.)

$ docker inspect --format '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}'  container_name

그외의 경우는 아래와 같이 확인

$ docker inspect --format '{{ .NetworkSettings.IPAddress }}' container_name

하나의 서버가 http와 ws스펙을 지원해야 하고, apache / nginx 같은 웹서버 뒷단(Proxy 환경)에서 서비스를 해야 되는 경우가 생겼다.

nginx같은 경우 간단하게 헤더 몇개만 추가해주면 자연스럽게 해결이 되는데,

apache같은 경우는 rewrite engine과 proxy_wstunnel engine을 enable 시킨 후, 별도의 조건까지 등록을 해주어야 가능하다.

아래 커맨드는 Ubuntu / Debian 계열 Linux 기준의 예제이고, 위에 얘기한 engine들을 enable 시키는 방법은 아래와 같다. 

$ a2enmod rewrite        # 보통 default로 enable 되어있는 경우가 많다.
$ a2enmod proxy_wstunnel

이제 Rewrite Engine 설정을 해준다. (포트가 6050인 경우)

# http-vhosts.conf
RewriteEngine on
RewriteCond %{HTTP:UPGRADE} =websocket [NC]
RewriteRule .* ws://127.0.0.1:6050/$1 [P,L]

ProxyPass / http://127.0.0.1:6050
ProxyPassReverse / http://127.0.0.1:6050

모든 작업 후 apache2를 재시작 해주면 적용된다.

정리하는 김에.. Nginx같은 경우는 아래와 같이 설정하면 됨.

server:
  location /:
    proxy_pass http://127.0.0.1:6050;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;