prometheus远程读接口实现对接clickhouse

remote_write:
  - url: "http://localhost:9201/write"
remote_read:
#  - url: "http://localhost:9201/read"
  - url: "http://192.168.2.133:31636/read"

package com.chinasoft;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.Array;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.springframework.stereotype.Controller;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.xerial.snappy.Snappy;
import org.xerial.snappy.SnappyInputStream;

import prometheus.Remote;
import prometheus.Remote.Query;
import prometheus.Remote.QueryResult;
import prometheus.Remote.ReadRequest;
import prometheus.Remote.ReadResponse;
import prometheus.Types.Label;
import prometheus.Types.LabelMatcher;
import prometheus.Types.Sample;
import prometheus.Types.TimeSeries;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

@Controller
public class DBController {

	@RequestMapping("/read")
	public void readDB(HttpServletRequest request, HttpServletResponse response) {
		printHeaders(request, response);
		try {
			ReadRequest readRequest = readParam(request);
			List<Query> queryList = readRequest.getQueriesList();
			for (Query query : queryList) {
				System.out.println("startTime == " + query.getStartTimestampMs());
				System.out.println("endTime == " + query.getEndTimestampMs());
				String name = "";
				for (LabelMatcher labelMatcher : query.getMatchersList()) {
					name = labelMatcher.getValue();
					System.out.println("matcher value==" + labelMatcher.getName());
					System.out.println("matcher type value==" + labelMatcher.getTypeValue());
					System.out.println("matcher type==" + labelMatcher.getType());
					System.out.println("matcher value==" + labelMatcher.getValue());
				}
//				ReadResponse readResponse = queryBySql(name, query.getStartTimestampMs(), query.getEndTimestampMs());
				ReadResponse readResponse = queryBySqlNew(query);
				writeResponse(readResponse, response);
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void writeResponse(ReadResponse readResponse, HttpServletResponse response) throws IOException {
		byte[] datas = readResponse.toByteArray();
		ServletOutputStream outStream = response.getOutputStream();
		datas = Snappy.compress(datas);
		response.setHeader("content-length", "" + datas.length);
		outStream.write(datas);
		outStream.flush();
		outStream.close();
	}

	private ReadRequest readParam(HttpServletRequest request) throws IOException {
		SnappyInputStream stream = new SnappyInputStream(request.getInputStream());
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		byte[] data = new byte[1024];
		int len = 0;
		while ((len = stream.read(data)) > -1) {
			baos.write(data, 0, len);
		}
		stream.close();
		ReadRequest readRequest = Remote.ReadRequest.parseFrom(baos.toByteArray());
		return readRequest;
	}

	private void printHeaders(HttpServletRequest request, HttpServletResponse response) {
		Enumeration<String> headers = request.getHeaderNames();
		for (; headers.hasMoreElements();) {
			String header = headers.nextElement();
			System.out.println("header:" + header + "==" + request.getHeader(header));
			response.setHeader(header, request.getHeader(header));
		}
//		Enumeration<String> attributeNames = request.getAttributeNames();
//		for (; attributeNames.hasMoreElements();) {
//			String attributeName = attributeNames.nextElement();
//			System.out.println("attributeName:" + attributeName + "==" + request.getAttribute(attributeName));
//		}
		Map<String, String[]> parameterMap = request.getParameterMap();
		for (Entry<String, String[]> entry : parameterMap.entrySet()) {
			System.out.println("Parameters:" + entry.getKey() + ",value==" + Arrays.asList(entry.getValue()));

		}
		System.out.println("url:" + request.getRequestURL().toString());
		System.out.println("ip:" + request.getRemoteHost());
	}

	private ReadResponse queryBySqlNew(Query query) {
		ReadResponse readResponse = Remote.ReadResponse.newBuilder().build();
		QueryResult queryResult = Remote.QueryResult.newBuilder().build();
		readResponse.setResultsList(Arrays.asList(queryResult));
		String sql = buildSql(query);
		try {
			Connection connection = getConn();
			Statement statement = connection.createStatement();
			System.out.println(sql);
			ResultSet rs = statement.executeQuery(sql);
			Map<String, TimeSeries> timeSerieMap = new HashMap<String, TimeSeries>();
			while (rs.next()) {
				String tagStrkey = rs.getString("tags");
				TimeSeries timeSeries = null;
				if (!timeSerieMap.containsKey(tagStrkey)) {
					timeSeries = TimeSeries.newBuilder().build();
					timeSerieMap.put(tagStrkey, timeSeries);
					List<Sample> sampleList = new ArrayList<Sample>();
					List<Label> labelList = new ArrayList<Label>();
					Array array = rs.getArray("tags");
					setLabelList(array, labelList);
					timeSeries.setLabelsList(labelList);
					timeSeries.setSamplesList(sampleList);
				} else {
					timeSeries = timeSerieMap.get(tagStrkey);
				}
				System.out.println(rs.getTimestamp(2).getTime());
				System.out.println(rs.getDouble(1));
				Sample sample = Sample.newBuilder().build();
				timeSeries.getSamplesList()
						.add(sample.setTimestamp(rs.getTimestamp("ts").getTime()).setValue(rs.getDouble("avg")));
			}

			queryResult.setTimeseriesList(new ArrayList<TimeSeries>(timeSerieMap.values()));
		} catch (Exception e) {
			e.printStackTrace();
		}
		return readResponse;
	}

	private void setLabelList(Array array, List<Label> labelList) {
		try {
			String[] arrays = (String[]) array.getArray();
			for (int j = 1; j < arrays.length; j++) {
				String str = arrays[j];
				String[] strs = str.split("=");
				labelList.add(Label.newBuilder().build().setName(strs[0]).setValue(strs[1]));
			}
			String[] first = arrays[0].split("=");
			labelList.add(Label.newBuilder().build().setName(first[0]).setValue(first[1]));
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	private String buildSql(Query query) {
		StringBuilder sb = new StringBuilder("select avg,ts,tags from metrics.samplesnew where");
		sb.append(" ts>=toDateTime(");
		sb.append(query.getStartTimestampMs() / 1000);
		sb.append(") and ts<=toDateTime(");
		sb.append(query.getEndTimestampMs() / 1000);
		sb.append(")");
		for (LabelMatcher labelMatcher : query.getMatchersList()) {
			System.out.println("matcher value==" + labelMatcher.getName());
			System.out.println("matcher type value==" + labelMatcher.getTypeValue());
			System.out.println("matcher type==" + labelMatcher.getType());
			System.out.println("matcher value==" + labelMatcher.getValue());
			if ("__name__".equals(labelMatcher.getName())) {
				appendNameWhere(labelMatcher, sb);
			} else {
				appendTagsWhere(labelMatcher, sb);
			}
		}
		sb.append(" order by ts");
		return sb.toString();
	}

	private void appendNameWhere(LabelMatcher labelMatcher, StringBuilder sb) {
		String tagV = labelMatcher.getValue();
		tagV = tagV.replaceAll("‘", "\\‘");
		if (StringUtils.isEmpty(tagV)) {
			return;
		}
		switch (labelMatcher.getTypeValue()) {
		case LabelMatcher.Type.EQ_VALUE:
			sb.append(" and name=‘");
			sb.append(tagV);
			sb.append("‘");
			break;
		case LabelMatcher.Type.NEQ_VALUE:
			sb.append(" and name!=‘");
			sb.append(tagV);
			sb.append("‘");
			break;
		case LabelMatcher.Type.RE_VALUE:
			sb.append(" and match(name, ‘");
			sb.append(tagV.replaceAll("/", "\\/"));
			sb.append("‘) = 1");
			break;
		case LabelMatcher.Type.NRE_VALUE:
			sb.append(" and match(name, ‘");
			sb.append(tagV.replaceAll("/", "\\/"));
			sb.append("‘) = 0");
			break;
		default:
			sb.append(" and name=‘");
			sb.append(tagV);
			sb.append("‘");
		}
	}

	private void appendTagsWhere(LabelMatcher labelMatcher, StringBuilder sb) {
		String tagK = labelMatcher.getName();
		String tagV = labelMatcher.getValue();
		tagV = tagV.replaceAll("‘", "\\‘");
		if (StringUtils.isEmpty(tagV)) {
			return;
		}
		// 等于 arrayExists(x -> x IN (%s), tags) = 1
		// 正则匹配 arrayExists(x -> 1 == match(x, ‘^%s=%s‘),tags) = 1
		switch (labelMatcher.getTypeValue()) {
		case LabelMatcher.Type.EQ_VALUE:
			sb.append(" and arrayExists(x -> x IN (‘");
			sb.append(tagK);
			sb.append("=");
			sb.append(tagV);
			sb.append("‘), tags) = 1");
			break;
		case LabelMatcher.Type.NEQ_VALUE:
			sb.append(" and arrayExists(x -> x IN (‘");
			sb.append(tagK);
			sb.append("=");
			sb.append(tagV);
			sb.append("‘), tags) = 0");
			break;
		case LabelMatcher.Type.RE_VALUE:
			sb.append(" and arrayExists(x -> 1 == match(x, ‘^");
			sb.append(tagK);
			sb.append("=");
			sb.append(tagV.replaceAll("/", "\\/"));
			sb.append("‘),tags) = 1");
			break;
		case LabelMatcher.Type.NRE_VALUE:
			sb.append(" and arrayExists(x -> 1 == match(x, ‘^");
			sb.append(tagK);
			sb.append("=");
			sb.append(tagV.replaceAll("/", "\\/"));
			sb.append("‘),tags) = 0");
			break;
		default:
			sb.append(" and arrayExists(x -> x IN (‘");
			sb.append(tagK);
			sb.append("=");
			sb.append(tagV);
			sb.append("‘), tags) = 1");
		}
	}

	private ReadResponse queryBySql(String name, long start, long end) {
		try {
			Connection connection = getConn();
			Statement statement = connection.createStatement();
			String sql = "select avg,ts,tags from metrics.samplesnew where name=‘" + name
					+ "‘ and toUnixTimestamp(ts)>=" + start / 1000 + " and toUnixTimestamp(ts)<=" + end / 1000
					+ " order by ts";
			ResultSet rs = statement.executeQuery(sql);
			System.out.println(sql);
			QueryResult queryResult = Remote.QueryResult.newBuilder().build();
			TimeSeries timeSeries = TimeSeries.newBuilder().build();
			queryResult.setTimeseriesList(Arrays.asList(timeSeries));
			ReadResponse readResponse = Remote.ReadResponse.newBuilder().build();
			readResponse.setResultsList(Arrays.asList(queryResult));
			List<Sample> sampleList = new ArrayList<Sample>();
			List<Label> labelList = new ArrayList<Label>();
			timeSeries.setLabelsList(labelList);
			timeSeries.setSamplesList(sampleList);
			Array array = null;
			while (rs.next()) {
				System.out.println(rs.getTimestamp(2).getTime());
				System.out.println(rs.getDouble(1));
				Sample sample = Sample.newBuilder().build();
				sampleList.add(sample.setTimestamp(rs.getTimestamp("ts").getTime()).setValue(rs.getDouble("avg")));
				array = rs.getArray(3);
			}
			if (array != null) {
				String[] arrays = (String[]) array.getArray();
				int i = 0;
				for (String str : arrays) {
					if (i == 0) {
						i++;
						continue;
					}
					String[] strs = str.split("=");
					labelList.add(Label.newBuilder().build().setName(strs[0]).setValue(strs[1]));
				}
				String[] first = arrays[0].split("=");
				labelList.add(Label.newBuilder().build().setName(first[0]).setValue(first[1]));
			}
			return readResponse;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	private static String clickhouseAddress = "jdbc:clickhouse://117.78.23.187:32123";

	private static String clickhouseUsername = "root";

	private static String clickhousePassword = "123456";

	private static String clickhouseDB = "metrics";

	private static Integer clickhouseSocketTimeout = 600000;

	private ClickHouseDataSource clickHouseDataSource;

	{

		ClickHouseProperties properties = new ClickHouseProperties();
		properties.setUser(clickhouseUsername);
		properties.setPassword(clickhousePassword);
		properties.setDatabase(clickhouseDB);
		properties.setSocketTimeout(clickhouseSocketTimeout);
		properties.setConnectionTimeout(60000);
		clickHouseDataSource = new ClickHouseDataSource(clickhouseAddress, properties);
	}

	private Connection getConn() {
		try {
			return clickHouseDataSource.getConnection();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

}