엘라스틱 서치 연동하기

스프링내에 발생하는 수많은 로그를 즉시 수집하고 엘라서틱서치와 연동하기 위한 방법

엘라스틱서치를 기본적으로 설치및 사용가능하고 java에서 logback의 로그기능 셋팅을 할수 있어야합니다. 6.2버젼에서 작동가능확인되었습니다.

목표:

기존 사용하는 로그(LogBack)기능을 사용하여 , 폴링방식이아닌 실시간 방식으로 연동
LogBack의 SocketAppend가 사용됨
집계를 위해서 보고되는 필드처리가 Json화되어여함

LogStash셋팅

logstash.conf

# [Beats input plugin]
# listen on port 5044 for incoming Beats connections
input {
  tcp {
    port => 5044
    codec => json_lines    
  }
}

filter {
  json {
    source => "message"
  }
}


# [Elasticsearch output plugin]
# index into Elasticsearch
output {
   elasticsearch {
     hosts => "localhost:9200"
     manage_template => false
     index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" 
   }
   stdout { codec => rubydebug }
 }

5044 tcp 포트를 통해, json 로그를 실시간으로 수집처리를 하겠다란 설정입니다.

의존Lib 셋팅

	 <!-- LogBack for LogStash -->
		<dependency>
            <groupId>net.logstash.logback</groupId>
            <artifactId>logstash-logback-encoder</artifactId>
            <version>5.1</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
         </dependency>  
         
         <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-access</artifactId>                
         </dependency>

logback의 logstash-logback-encoder가 엘라스틱이 이해할수 있는 형태의 로그로 만들어줍니다.

LogBack.xml 설정

<?xml version="1.0" encoding="UTF-8"?>
<configuration>  
 <appender name="STASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
      <keepAliveDuration>5 minutes</keepAliveDuration>
      <reconnectionDelay>10 second</reconnectionDelay>
      <waitStrategyType>sleeping</waitStrategyType>
      <writeBufferSize>16384</writeBufferSize>
      <destination>180.70.98.37:5044</destination>      
	 <encoder class="net.logstash.logback.encoder.LogstashEncoder">
	 	<version>6</version>	 	  	 	  	
	 </encoder>	 	 	
  </appender>


  <logger name="org.springframework.data" level="error" />


  <root level="info">
    <appender-ref ref="STASH"/>
  </root>
</configuration>

로그기능은 파일및 DB또는 콘솔등 다양한 아웃풋지정이 가능하며
여기서는 LogStash Socket 전송기능을 설정을 하였습니다.
몇가지 LogsStash가 띄워진 서버정보가 설정이됩니다.

Log에 사용되는 Class

import java.io.Serializable;
import com.fasterxml.jackson.databind.ObjectMapper;


public class InFlowLog implements Serializable {
    // 로그를 정의한다. 로그를 귀찮게 필드명을 왜 지정을하느냐?
    // RDB에 저장하지 않고도 Table과 같이,키바나를 통해 SQL문을 사용해서 조회및 집계가가능하며
    // 궁극적으로 개발자가 아닌사람도... SQL문 사용없이 클릭만으로 리포팅생성이 가능해집니다.


	private String category1;

	private String category2;

	private String category3;

	private String category4;

	private String region;

	private String goodscode;
	
	private String goodsname;
	
	private String search;
	
	private String router;

	private String action;
	
	private long ctime;		
	
	private String userip;
	
	private int age;
	
	private String sex;
	
	private String ordertip;
	
	private String refgoodscode;
	
	@Override
    public String toString() {
		ObjectMapper mapper = new ObjectMapper();
		String jsonString ="";
		try {
			jsonString = mapper.writeValueAsString(this);			
		} catch(IOException e) {
			jsonString = "jsonerror";			
		}
		return jsonString;
	}

}

로그를 구조를 정의할수 있는 가장 간단한 방법입니다.
다양한 로그에 필요한 멤버정의후, toString만 재정의하여 JsonString으로 표현하게 설정을 합니다.

Log에 사용되는 Class

InFlowLog addLog = new InFlowLog();
logger.info( "{}",addLog.toString()  );	//이것이 의미가 있는 로그....

로그를 쌓거나 전송하는 가장 일반적인 방법이며, 우리가 원하는 로그정보를 채운후
전송을 하면됩니다.

연동결과 확인

우리가 정의한 로그 Class의 멤버 필드가, 정확하게 엘라스틱이 이해하는 필드명이 되면 성공입니다.
이것이 중요한 이유는, 추후 이렇게 정의된 필드를 통해서 다양한 집계처리및 리포팅이 가능하기때문입니다.

신규 로그 필드가 추가되었기때문에, 인덱스 패턴을 추가해야 하며 기존 인덱스를 Refresh하여 추가하거나, 신규로 추가합니다.
RDB에 저장한 경우 엔티티수정이 필요하면 DBA에게 허락을 맡아야하고 , 필드 변경에따라 어떠한 오동작을 할지 예측할수 없으나
엘라서치 내에 엔티티수정은 , 알아서 감지하고 추가합니다. NOSQL의 장점이기도하며 로그시스템에 더 적합함을 의미합니다.

GEOPINT 연동

위경도 로그 구조

public class Location implements Serializable {

	private Double lat;

	private Double lon;
	

	public Double getLat() {
		return lat;
	}

	public void setLat(Double lat) {
		this.lat = lat;
	}

	public Double getLon() {
		return lon;
	}

	public void setLon(Double lon) {
		this.lon = lon;
	}

}


public class InFlowLog implements Serializable {
  private Location location;
}

키바나에서 geopoint 맵핑명령실행

put logstash-idxname
{
    "mappings":{
      "doc":{
              "properties": {
              "location": {
                "type": "geo_point"
              }
            }
      }
    }
}

맵핑설정을 먼저하는게 좋다.

성공시

Links:

Content

Space Tools

LogStash셋팅

의존Lib 셋팅

LogBack.xml 설정

Log에 사용되는 Class

Log에 사용되는 Class

연동결과 확인

GEOPINT 연동

위경도 로그 구조

키바나에서 geopoint 맵핑명령실행

성공시

3 Comments

Anonymous

Anonymous

Anonymous