@@ -258,7 +266,7 @@ const DashboardDragChart = (props: PropsType): JSX.Element => {
}}
>
{metricChartData.map((data) => {
- const { metricName, metricUnit, metricLines } = data;
+ const { metricName, metricUnit, metricLines, showLegend } = data;
return (
@@ -301,7 +309,7 @@ const DashboardDragChart = (props: PropsType): JSX.Element => {
eventBus={busInstance}
propChartData={metricLines}
optionMergeProps={{ replaceMerge: curHeaderOptions.isAutoReload ? ['xAxis'] : ['series'] }}
- {...getChartConfig(`${metricName}{unit|（${metricUnit}）}`, metricLines.length)}
+ {...getChartConfig(`${metricName}{unit|（${metricUnit}）}`, metricLines.length, showLegend)}
/>
);
@@ -321,4 +329,4 @@ const DashboardDragChart = (props: PropsType): JSX.Element => {
);
};
-export default DashboardDragChart;
+export default DraggableCharts;
diff --git a/km-console/packages/layout-clusters-fe/src/components/Message/index.tsx b/km-console/packages/layout-clusters-fe/src/components/Message/index.tsx
new file mode 100644
index 00000000..5791939c
--- /dev/null
+++ b/km-console/packages/layout-clusters-fe/src/components/Message/index.tsx
@@ -0,0 +1,54 @@
+import React from 'react';
+import { IconFont } from '@knowdesign/icons';
+import { message } from 'knowdesign';
+import { ArgsProps, ConfigOnClose } from 'knowdesign/es/basic/message';
+
+type ConfigContent = React.ReactNode;
+type ConfigDuration = number | (() => void);
+type JointContent = ConfigContent | ArgsProps;
+
+message.config({
+ top: 16,
+});
+
+function isArgsProps(content: JointContent): content is ArgsProps {
+ return Object.prototype.toString.call(content) === '[object Object]' && !!(content as ArgsProps).content;
+}
+
+const openMessage = (
+ type: 'info' | 'success' | 'warning' | 'error',
+ content: JointContent,
+ duration?: ConfigDuration,
+ onClose?: ConfigOnClose
+) => {
+ if (isArgsProps(content)) {
+ message[type]({
+ icon:
,
+ ...content,
+ });
+ } else {
+ message[type]({
+ icon:
,
+ content,
+ duration,
+ onClose,
+ });
+ }
+};
+
+const customMessage = {
+ info(content: JointContent, duration?: ConfigDuration, onClose?: ConfigOnClose) {
+ openMessage('info', content, duration, onClose);
+ },
+ success(content: JointContent, duration?: ConfigDuration, onClose?: ConfigOnClose) {
+ openMessage('success', content, duration, onClose);
+ },
+ warning(content: JointContent, duration?: ConfigDuration, onClose?: ConfigOnClose) {
+ openMessage('warning', content, duration, onClose);
+ },
+ error(content: JointContent, duration?: ConfigDuration, onClose?: ConfigOnClose) {
+ openMessage('error', content, duration, onClose);
+ },
+};
+
+export default customMessage;
diff --git a/km-console/packages/layout-clusters-fe/src/components/Notification/index.tsx b/km-console/packages/layout-clusters-fe/src/components/Notification/index.tsx
new file mode 100644
index 00000000..9673d511
--- /dev/null
+++ b/km-console/packages/layout-clusters-fe/src/components/Notification/index.tsx
@@ -0,0 +1,33 @@
+import React from 'react';
+import { notification } from 'knowdesign';
+import { ArgsProps } from 'knowdesign/es/basic/notification';
+import { IconFont } from '@knowdesign/icons';
+
+notification.config({
+ top: 16,
+ duration: 3,
+});
+
+const open = (type: 'info' | 'success' | 'warning' | 'error', content: ArgsProps) => {
+ notification[type]({
+ icon:
,
+ ...content,
+ });
+};
+
+const customNotification = {
+ info(content: ArgsProps) {
+ open('info', content);
+ },
+ success(content: ArgsProps) {
+ open('success', content);
+ },
+ warning(content: ArgsProps) {
+ open('warning', content);
+ },
+ error(content: ArgsProps) {
+ open('error', content);
+ },
+};
+
+export default customNotification;
diff --git a/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaChange.tsx b/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaChange.tsx
index 16637b9c..cd4f05ba 100644
--- a/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaChange.tsx
+++ b/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaChange.tsx
@@ -15,10 +15,10 @@ import {
Tag,
Utils,
AppContainer,
- message,
Divider,
Space,
} from 'knowdesign';
+import message from '@src/components/Message';
import './index.less';
import Api, { MetricType } from '@src/api/index';
import moment from 'moment';
diff --git a/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaMove.tsx b/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaMove.tsx
index a6089e80..a93b26ef 100644
--- a/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaMove.tsx
+++ b/km-console/packages/layout-clusters-fe/src/components/TopicJob/ReplicaMove.tsx
@@ -14,12 +14,12 @@ import {
Table,
Utils,
AppContainer,
- message,
Space,
Divider,
Transfer,
Tooltip,
} from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import './index.less';
import Api, { MetricType } from '@src/api/index';
diff --git a/km-console/packages/layout-clusters-fe/src/constants/axiosConfig.ts b/km-console/packages/layout-clusters-fe/src/constants/axiosConfig.ts
index fb0ddf44..5a087ca0 100644
--- a/km-console/packages/layout-clusters-fe/src/constants/axiosConfig.ts
+++ b/km-console/packages/layout-clusters-fe/src/constants/axiosConfig.ts
@@ -1,4 +1,6 @@
-import { notification, Utils } from 'knowdesign';
+import { Utils } from 'knowdesign';
+import notification from '@src/components/Notification';
+
const { EventBus } = Utils;
export const licenseEventBus = new EventBus();
@@ -42,8 +44,8 @@ serviceInstance.interceptors.response.use(
licenseEventBus.emit('licenseError', desc);
} else {
notification.error({
- message: desc,
- duration: 3,
+ message: '错误信息',
+ description: desc,
});
}
throw res;
@@ -83,36 +85,31 @@ const dealResponse = (error: any) => {
case 405:
notification.error({
message: '错误',
- duration: 3,
description: `${error.response.data.message || '请求方式错误'}`,
});
break;
case 500:
notification.error({
message: '错误',
- duration: 3,
description: '服务错误，请重试！',
});
break;
case 502:
notification.error({
message: '错误',
- duration: 3,
description: '网络错误，请重试！',
});
break;
default:
notification.error({
message: '连接出错',
- duration: 3,
description: `${error.response.status}`,
});
}
} else {
notification.error({
+ message: '连接超时!',
description: '请重试或检查服务',
- message: '连接超时! ',
- duration: 3,
});
}
return Promise.reject(error);
diff --git a/km-console/packages/layout-clusters-fe/src/constants/chartConfig.ts b/km-console/packages/layout-clusters-fe/src/constants/chartConfig.ts
index 955bcfa0..f566e850 100644
--- a/km-console/packages/layout-clusters-fe/src/constants/chartConfig.ts
+++ b/km-console/packages/layout-clusters-fe/src/constants/chartConfig.ts
@@ -12,30 +12,38 @@ export interface MetricInfo {
}
// 接口返回图表原始数据类型
-export interface MetricDefaultChartDataType {
+export interface OriginMetricData {
metricName: string;
- metricLines: {
+ metricLines?: {
name: string;
createTime: number;
updateTime: number;
metricPoints: {
aggType: string;
timeStamp: number;
- value: number;
+ value: string;
createTime: number;
updateTime: number;
}[];
}[];
+ metricPoints?: {
+ aggType: string;
+ name: string;
+ timeStamp: number;
+ value: string;
+ }[];
}
// 格式化后图表数据类型
-export interface MetricChartDataType {
+export interface FormattedMetricData {
metricName: string;
metricUnit: string;
metricLines: {
name: string;
data: (string | number)[][];
}[];
+ showLegend: boolean;
+ targetUnit: [string, number] | undefined;
}
// 图表颜色库
@@ -55,22 +63,38 @@ export const CHART_COLOR_LIST = [
'#C9E795',
];
-// 图表存储单位换算
-export const UNIT_MAP = {
+// 图表存储单位
+export const MEMORY_MAP = {
TB: Math.pow(1024, 4),
GB: Math.pow(1024, 3),
MB: Math.pow(1024, 2),
KB: 1024,
};
-export const getUnit = (value: number) => Object.entries(UNIT_MAP).find(([, size]) => value / size >= 1) || ['Byte', 1];
-
-// 图表数字单位换算
-export const DATA_NUMBER_MAP = {
+// 图表时间单位
+export const TIME_MAP = {
+ h: 1000 * 60 * 60,
+ min: 1000 * 60,
+ s: 1000,
+};
+// 图表数字单位
+export const NUM_MAP = {
十亿: Math.pow(1000, 3),
百万: Math.pow(1000, 2),
千: 1000,
};
-export const getDataNumberUnit = (value: number) => Object.entries(DATA_NUMBER_MAP).find(([, size]) => value / size >= 1) || ['', 1];
+const calculateUnit = (map: { [unit: string]: number }, targetValue: number) => {
+ return Object.entries(map).find(([, size]) => targetValue / size >= 1);
+};
+const getMemoryUnit = (value: number) => calculateUnit(MEMORY_MAP, value) || ['Byte', 1];
+const getTimeUnit = (value: number) => calculateUnit(TIME_MAP, value) || ['ms', 1];
+const getNumUnit = (value: number) => calculateUnit(NUM_MAP, value) || ['', 1];
+
+export const getDataUnit = {
+ Memory: getMemoryUnit,
+ Time: getTimeUnit,
+ Num: getNumUnit,
+};
+export type DataUnitType = keyof typeof getDataUnit;
// 图表补点间隔计算
export const SUPPLEMENTARY_INTERVAL_MAP = {
@@ -112,9 +136,9 @@ export const resolveMetricsRank = (metricList: MetricInfo[]) => {
};
};
-// 补点
+// 图表补点
export const supplementaryPoints = (
- lines: MetricChartDataType['metricLines'],
+ lines: FormattedMetricData['metricLines'],
timeRange: readonly [number, number],
extraCallback?: (point: [number, 0]) => any[]
): void => {
@@ -165,19 +189,39 @@ export const supplementaryPoints = (
// 格式化图表数据
export const formatChartData = (
// 图表源数据
- metricData: MetricDefaultChartDataType[],
+ metricsData: OriginMetricData[],
// 获取指标单位
getMetricDefine: (type: MetricType, metric: string) => MetricsDefine[keyof MetricsDefine],
// 指标类型
metricType: MetricType,
// 图表时间范围，用于补点
timeRange: readonly [number, number],
- transformUnit: [string, number] = undefined
-): MetricChartDataType[] => {
- return metricData.map(({ metricName, metricLines }) => {
+ targetUnit: [string, number] = undefined
+): FormattedMetricData[] => {
+ return metricsData.map((originData) => {
+ const { metricName } = originData;
const curMetricInfo = (getMetricDefine && getMetricDefine(metricType, metricName)) || null;
- const isByteUnit = curMetricInfo?.unit?.toLowerCase().includes('byte');
+ let showLegend = true;
+ let metricLines = [];
let maxValue = -1;
+ let unitType: DataUnitType;
+
+ if (originData?.metricLines && originData?.metricLines !== null) {
+ metricLines = originData.metricLines;
+ } else {
+ showLegend = false;
+ metricLines = [
+ {
+ name: metricName,
+ metricPoints: originData.metricPoints,
+ },
+ ];
+ }
+
+ {
+ const originUnit = curMetricInfo?.unit?.toLowerCase();
+ unitType = originUnit.includes('byte') ? 'Memory' : originUnit.includes('ms') ? 'Time' : 'Num';
+ }
const PointsMapMethod = ({ timeStamp, value }: { timeStamp: number; value: string | number }) => {
let parsedValue: string | number = Number(value);
@@ -194,7 +238,7 @@ export const formatChartData = (
};
// 初始化返回结构
- const chartData = {
+ const chartData: FormattedMetricData = {
metricName,
metricUnit: curMetricInfo?.unit || '',
metricLines: metricLines
@@ -203,17 +247,21 @@ export const formatChartData = (
name,
data: metricPoints.map(PointsMapMethod),
})),
+ showLegend,
+ targetUnit: undefined,
};
// 按时间先后进行对图表点排序
chartData.metricLines.forEach(({ data }) => data.sort((a, b) => (a[0] as number) - (b[0] as number)));
// 图表值单位转换
if (maxValue > 0) {
- const [unitName, unitSize]: [string, number] = transformUnit || isByteUnit ? getUnit(maxValue) : getDataNumberUnit(maxValue);
- chartData.metricUnit = isByteUnit
- ? chartData.metricUnit.toLowerCase().replace('byte', unitName)
- : `${unitName}${chartData.metricUnit}`;
- chartData.metricLines.forEach(({ data }) => data.forEach((point: any) => (point[1] /= unitSize)));
+ const [unitName, unitSize]: [string, number] = targetUnit || getDataUnit[unitType](maxValue);
+ chartData.targetUnit = [unitName, unitSize];
+ chartData.metricUnit =
+ unitType !== 'Num'
+ ? chartData.metricUnit.toLowerCase().replace(unitType === 'Memory' ? 'byte' : 'ms', unitName)
+ : `${unitName}${chartData.metricUnit}`;
+ chartData.metricLines.forEach(({ data }) => data.forEach((point: any) => parseFloat((point[1] /= unitSize).toFixed(3))));
}
// 补点
@@ -231,7 +279,7 @@ const tooltipFormatter = (date: any, arr: any, tooltip: any) => {
-
+
${item.seriesName}
@@ -324,6 +372,7 @@ export const getBasicChartConfig = (props: any = {}) => {
tooltip: false,
...legend,
},
+ color: CHART_COLOR_LIST,
// 横坐标配置
xAxis: {
type: 'category',
diff --git a/km-console/packages/layout-clusters-fe/src/pages/BrokerDashboard/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/BrokerDashboard/index.tsx
index a3398dec..a026d43f 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/BrokerDashboard/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/BrokerDashboard/index.tsx
@@ -1,7 +1,7 @@
import React from 'react';
import { MetricType } from '@src/api';
import BrokerHealthCheck from '@src/components/CardBar/BrokerHealthCheck';
-import DashboardDragChart from '@src/components/DashboardDragChart';
+import DraggableCharts from '@src/components/DraggableCharts';
import DBreadcrumb from 'knowdesign/es/extend/d-breadcrumb';
import { AppContainer } from 'knowdesign';
@@ -19,7 +19,7 @@ const BrokerDashboard = (): JSX.Element => {
/>
-
+
);
};
diff --git a/km-console/packages/layout-clusters-fe/src/pages/BrokerDetail/ConfigurationEdit.tsx b/km-console/packages/layout-clusters-fe/src/pages/BrokerDetail/ConfigurationEdit.tsx
index a362acc0..f4fda39a 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/BrokerDetail/ConfigurationEdit.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/BrokerDetail/ConfigurationEdit.tsx
@@ -1,5 +1,6 @@
import React, { useEffect } from 'react';
-import { Drawer, Form, Input, Space, Button, Checkbox, Utils, Row, Col, Divider, message } from 'knowdesign';
+import { Drawer, Form, Input, Space, Button, Checkbox, Utils, Row, Col, Divider } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import { useParams } from 'react-router-dom';
import Api from '@src/api';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/BrokerList/config.tsx b/km-console/packages/layout-clusters-fe/src/pages/BrokerList/config.tsx
index a0f6cbe5..d2b28cda 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/BrokerList/config.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/BrokerList/config.tsx
@@ -3,6 +3,7 @@ import React from 'react';
import { timeFormat, getSizeAndUnit } from '../../constants/common';
import moment from 'moment';
import { Tag, Tooltip } from 'knowdesign';
+import { IconFont } from '@knowdesign/icons';
export const getBrokerListColumns = (arg?: any) => {
const columns = [
@@ -82,6 +83,14 @@ export const getBrokerListColumns = (arg?: any) => {
dataIndex: 'jmxPort',
key: 'jmxPort',
width: 100,
+ render: (t: string, r: any) => {
+ return (
+
+
+ {t}
+
+ );
+ },
},
{
title: 'Partitions',
diff --git a/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/Detail.tsx b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/Detail.tsx
new file mode 100644
index 00000000..34001fa0
--- /dev/null
+++ b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/Detail.tsx
@@ -0,0 +1,249 @@
+import React, { useState, useEffect } from 'react';
+import { useParams, useHistory } from 'react-router-dom';
+import { Drawer, ProTable, Utils } from 'knowdesign';
+import { IconFont } from '@knowdesign/icons';
+import API from '@src/api/index';
+import { defaultPagination, hashDataParse } from '@src/constants/common';
+import { getGtoupTopicColumns } from './config';
+import { ExpandedRow } from './ExpandedRow';
+import ResetOffsetDrawer from './ResetOffsetDrawer';
+const { request } = Utils;
+
+export interface MetricLine {
+ createTime?: number;
+ metricPoints: Array<{
+ aggType: string;
+ createTime: number;
+ timeStamp: number;
+ unit: string;
+ updateTime: number;
+ value: number;
+ }>;
+ name: string;
+ updateTime?: number;
+}
+
+export interface MetricData {
+ metricLines?: Array
;
+ metricLine?: MetricLine;
+ metricName: string;
+}
+
+export interface HashData {
+ groupName: string;
+ topicName: string;
+}
+
+const metricConsts = ['LogEndOffset', 'OffsetConsumed', 'Lag'];
+const metricWithType = [
+ { metricName: 'LogEndOffset', metricType: 104 },
+ { metricName: 'OffsetConsumed', metricType: 102 },
+ { metricName: 'Lag', metricType: 102 },
+];
+
+const GroupDetail = (props: any) => {
+ const { scene } = props;
+ const urlParams = useParams<{
+ clusterId: string;
+ }>();
+ const now = Date.now();
+ const history = useHistory();
+ const [hashData, setHashData] = useState({ groupName: '', topicName: '' });
+ const [visible, setVisible] = useState(false);
+
+ const [topicData, setTopicData] = useState([]);
+ const [loading, setLoading] = useState(true);
+ const [pagination, setPagination] = useState(defaultPagination);
+ const [expandedData, setExpandedData] = useState([]);
+ const [chartData, setChartData] = useState>([]);
+ const [loadingObj, setLoadingObj] = useState({});
+ const [timeRange, setTimeRange] = useState([now - 24 * 60 * 60 * 1000, now]);
+ const [curPartition, setCurPartition] = useState('');
+ const [groupMetricsData, setGroupMetricsData] = useState>([]);
+ const [openKeys, setOpenKeys] = useState();
+ const [resetOffsetVisible, setResetOffsetVisible] = useState(false);
+ const [resetOffsetArg, setResetOffsetArg] = useState({});
+
+ const genData = async ({ pageNo, pageSize, groupName }: any) => {
+ if (urlParams?.clusterId === undefined) return;
+
+ setLoading(true);
+ const params = {
+ // searchKeywords: '',
+ pageNo,
+ pageSize,
+ };
+
+ request(API.getGroupTopicList(+urlParams?.clusterId, groupName), { params })
+ .then((res: any) => {
+ setVisible(true);
+ setPagination({
+ current: res.pagination?.pageNo,
+ pageSize: res.pagination?.pageSize,
+ total: res.pagination?.total,
+ });
+ const newTopicListData = res?.bizData.map((item: any) => {
+ return {
+ ...item,
+ key: item.topicName,
+ };
+ });
+ setTopicData(newTopicListData || []);
+ setLoading(false);
+ })
+ .catch((err) => {
+ setLoading(false);
+ });
+ };
+
+ const onClose = () => {
+ setVisible(false);
+ // clean hash'
+ scene === 'topicDetail' && history.goBack();
+ scene !== 'topicDetail' && window.history.pushState('', '', location.pathname);
+ };
+
+ const resetOffset = (record: any) => {
+ setResetOffsetVisible(true);
+ setResetOffsetArg({
+ topicName: record?.topicName,
+ groupName: record?.groupName,
+ });
+ };
+
+ const onTableChange = (pagination: any, filters: any, sorter: any) => {
+ genData({ pageNo: pagination.current, pageSize: pagination.pageSize, filters, sorter, groupName: hashData.groupName });
+ };
+
+ const onClickExpand = (expanded: any, record: any) => {
+ const key = record?.key;
+ // 之前展开过
+ if (expandedData[key]?.length) return;
+ // 第一次展开
+ setOpenKeys(key);
+ // const loading = { ...loadingObj };
+ // loading[key] = true;
+ // setLoadingObj(loading);
+ // expanded && queryExpandedData(record, key);
+ };
+
+ useEffect(() => {
+ const hashData = hashDataParse(location.hash);
+ if (!hashData.groupName) {
+ setVisible(false);
+ }
+ setHashData(hashData);
+ // 获取分区列表 为图表模式做准备
+ hashData.groupName && genData({ pageNo: 1, pageSize: pagination.pageSize, groupName: hashData.groupName });
+ // getConsumersMetadata(hashData).then((res: any) => {
+ // if (res.exist) {
+ // setVisible(false);
+ // history.push(`/cluster/${params?.clusterId}/consumers`);
+ // return;
+ // }
+ // setVisible(true);
+ // getTopicGroupPartitionsHistory(hashData)
+ // .then((data: any) => {
+ // if (data.length > 0) {
+ // setCurPartition(data[0].partition);
+ // }
+ // setPartitionList(data);
+ // return data;
+ // })
+ // .then((data) => {
+ // getTopicGroupMetricHistory(data, hashData);
+ // })
+ // .catch((e) => {
+ // history.push(`/cluster/${params?.clusterId}/consumers`);
+ // setVisible(false);
+ // });
+ // // 获取Consumer列表 表格模式
+ // getTopicGroupMetric(hashData);
+ // });
+ }, [hashDataParse(location.hash).groupName]);
+
+ return (
+
+ // {global.hasPermission &&
+ // global.hasPermission(
+ // scene === 'topicDetail' ? ClustersPermissionMap.TOPIC_RESET_OFFSET : ClustersPermissionMap.CONSUMERS_RESET_OFFSET
+ // ) && }
+ //
+ //
+ // }
+ >
+ (
+
+ ),
+ // expandedRowRender,
+ onExpand: onClickExpand,
+ columnWidth: '20px',
+ fixed: 'left',
+ expandIcon: ({ expanded, onExpand, record }: any) => {
+ return expanded ? (
+ {
+ onExpand(record, e);
+ }}
+ />
+ ) : (
+ {
+ onExpand(record, e);
+ }}
+ />
+ );
+ },
+ },
+ style: {
+ width: '1032px',
+ },
+ },
+ }}
+ />
+
+
+ );
+};
+
+export default GroupDetail;
diff --git a/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/ExpandedRow.tsx b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/ExpandedRow.tsx
new file mode 100644
index 00000000..fac1e99a
--- /dev/null
+++ b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/ExpandedRow.tsx
@@ -0,0 +1,413 @@
+import React, { useEffect, useState } from 'react';
+import { useParams, useHistory } from 'react-router-dom';
+import { ProTable, Utils, DRangeTime, Select, SingleChart } from 'knowdesign';
+import SwitchTab from '@src/components/SwitchTab';
+import { CHART_COLOR_LIST, getBasicChartConfig } from '@src/constants/chartConfig';
+import ContentWithCopy from '@src/components/CopyContent';
+import { IconFont } from '@knowdesign/icons';
+import API from '@src/api/index';
+import { hashDataParse } from '@src/constants/common';
+const { Option } = Select;
+
+export interface MetricLine {
+ createTime?: number;
+ metricPoints: Array<{
+ aggType: string;
+ createTime: number;
+ timeStamp: number;
+ unit: string;
+ updateTime: number;
+ value: number;
+ }>;
+ name: string;
+ updateTime?: number;
+}
+
+export interface MetricData {
+ metricLines?: Array;
+ metricLine?: MetricLine;
+ metricName: string;
+}
+
+export interface HashData {
+ groupName: string;
+ topicName: string;
+}
+
+const metricConsts = ['LogEndOffset', 'OffsetConsumed', 'Lag'];
+const metricWithType = [
+ { metricName: 'LogEndOffset', metricType: 104 },
+ { metricName: 'OffsetConsumed', metricType: 102 },
+ { metricName: 'Lag', metricType: 102 },
+];
+
+export const ExpandedRow: any = ({ record, groupName }: any) => {
+ const params: any = useParams<{
+ clusterId: string;
+ }>();
+ const history = useHistory();
+ const now = Date.now();
+ const [allGroupMetricsData, setAllGroupMetricsData] = useState>([]);
+ const [showMode, setShowMode] = useState('table');
+ const [curPartition, setCurPartition] = useState('');
+ const [partitionList, setPartitionList] = useState([]);
+ const [timeRange, setTimeRange] = useState([now - 24 * 60 * 60 * 1000, now]);
+ const [consumerListLoading, setConsumerListLoading] = useState(true);
+ const [consumerList, setConsumerList] = useState([]);
+ const [groupMetricsData, setGroupMetricsData] = useState>([]);
+ const clusterId = Number(params.clusterId);
+ const [pagination, setPagination] = useState({
+ current: 1,
+ pageSize: 5,
+ simple: true,
+ hideOnSinglePage: false,
+ });
+ const columns = [
+ {
+ title: 'Partition',
+ dataIndex: 'partitionId',
+ key: 'partitionId',
+ lineClampOne: true,
+ needTooltip: true,
+ width: 180,
+ },
+ {
+ title: 'Member ID',
+ dataIndex: 'memberId',
+ key: 'memberId',
+ width: 200,
+ render: (v: string) => {
+ return v ? : '-';
+ },
+ },
+ {
+ title: 'Current Offset',
+ dataIndex: 'OffsetConsumed',
+ key: 'OffsetConsumed',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.OffsetConsumed.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'OffsetConsumed' && item.metricType === 102)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'OffsetConsumed' && item.metricType === 102)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 1
+ // }
+ },
+ {
+ title: 'Log End Offset',
+ dataIndex: 'LogEndOffset',
+ key: 'LogEndOffset',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.LogEndOffset.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'LogEndOffset' && item.metricType === 104)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'LogEndOffset' && item.metricType === 104)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 2
+ // }
+ },
+ {
+ title: 'Lag',
+ dataIndex: 'Lag',
+ key: 'Lag',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.Lag.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'Lag' && item.metricType === 102)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'Lag' && item.metricType === 102)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 3
+ // }
+ },
+ {
+ title: 'Host',
+ dataIndex: 'host',
+ key: 'host',
+ },
+ {
+ title: 'Client ID',
+ dataIndex: 'clientId',
+ key: 'clientId',
+ needTooltip: true,
+ lineClampOne: true,
+ width: 200,
+ },
+ ];
+
+ const getTopicGroupMetric = ({ pagination = { current: 1, pageSize: 10 }, sorter = {} }: { pagination?: any; sorter?: any }) => {
+ setConsumerListLoading(true);
+ const params: any = {
+ // metricRealTimes: metricWithType,
+ latestMetricNames: metricConsts,
+ pageNo: pagination.current,
+ pageSize: pagination.pageSize,
+ sortField: sorter.field || undefined,
+ sortType: sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : undefined,
+ };
+
+ return Utils.post(
+ API.getTopicGroupMetric({
+ clusterId,
+ groupName: groupName,
+ topicName: record.topicName,
+ }),
+ params
+ )
+ .then((data: any) => {
+ if (!data) return;
+ setPagination((origin: any) => {
+ return { ...origin, current: data.pagination?.pageNo, pageSize: data.pagination?.pageSize };
+ });
+ setConsumerList(data?.bizData || []);
+ })
+ .finally(() => {
+ setConsumerListLoading(false);
+ });
+ };
+
+ const getTopicGroupMetricHistory = (partitions: Array) => {
+ const params = {
+ aggType: 'sum',
+ groupTopics: partitions.map((p: any) => ({
+ partition: p.value,
+ topic: record.topicName,
+ })),
+ group: groupName,
+ metricsNames: metricWithType.map((item) => item.metricName),
+ startTime: timeRange[0],
+ endTime: timeRange[1],
+ topNu: 0,
+ };
+ Utils.post(API.getTopicGroupMetricHistory(clusterId), params).then((data: Array) => {
+ // ! 替换接口返回
+ setAllGroupMetricsData(data);
+ });
+ };
+
+ const getConsumersMetadata = (hashData: HashData) => {
+ return Utils.request(API.getConsumersMetadata(params.clusterId, groupName, record.topicName));
+ };
+
+ const getTopicsMetaData = () => {
+ return Utils.request(API.getTopicsMetaData(record?.topicName, +params.clusterId));
+ };
+ const onTableChange = (pagination: any, filters: any, sorter: any) => {
+ getTopicGroupMetric({ pagination, sorter });
+ };
+
+ // useEffect(() => {
+ // getTopicGroupMetric();
+ // }, [sortObj]);
+
+ useEffect(() => {
+ const hashData = hashDataParse(location.hash);
+ // if (!hashData.groupName) return;
+ // 获取分区列表 为图表模式做准备
+
+ getConsumersMetadata(hashData).then((res: any) => {
+ if (!res.exist) {
+ history.push(`/cluster/${params?.clusterId}/consumers`);
+ return;
+ }
+ getTopicsMetaData()
+ .then((data: any) => {
+ const partitionLists = (data?.partitionIdList || []).map((item: any) => {
+ return {
+ label: item,
+ value: item,
+ };
+ });
+ setCurPartition(partitionLists?.[0]?.value);
+ setPartitionList(partitionLists);
+ getTopicGroupMetricHistory(partitionLists);
+ })
+ .catch((e) => {
+ history.push(`/cluster/${params?.clusterId}/consumers`);
+ });
+ // 获取Consumer列表 表格模式
+ getTopicGroupMetric({});
+ });
+ }, [hashDataParse(location.hash).groupName]);
+
+ useEffect(() => {
+ if (partitionList.length === 0) return;
+ getTopicGroupMetricHistory(partitionList);
+ }, [timeRange]);
+
+ useEffect(() => {
+ if (curPartition === '' || allGroupMetricsData.length === 0) return;
+ const filteredData = allGroupMetricsData.map((item) => {
+ const allData = item.metricLines.reduce(
+ (acc, cur) => {
+ if (acc.metricLine.metricPoints.length === 0) {
+ acc.metricLine.metricPoints = cur.metricPoints.map((p: any) => ({
+ timeStamp: p.timeStamp,
+ value: Number(p.value),
+ }));
+ } else {
+ acc.metricLine.metricPoints.forEach((mp) => {
+ const curMetricPoint = cur.metricPoints.find((curmp) => curmp.timeStamp === mp.timeStamp);
+ mp.value += curMetricPoint ? Number(curMetricPoint.value) : 0;
+ });
+ }
+ return acc;
+ },
+ {
+ metricName: item.metricName,
+ metricLine: {
+ name: 'all',
+ metricPoints: [],
+ },
+ }
+ );
+ return curPartition === '__all__'
+ ? allData
+ : {
+ metricName: item.metricName,
+ metricLine: item.metricLines.find((line) => line.name.indexOf(curPartition) >= 0),
+ };
+ });
+ setGroupMetricsData(filteredData);
+ }, [curPartition, allGroupMetricsData]);
+
+ return (
+
+
+
+
+
+ {showMode === 'chart' && (
+
+ )}
+ {showMode === 'chart' && (
+
{
+ setTimeRange(o);
+ }}
+ >
+ )}
+ {showMode === 'chart' &&
}
+
setShowMode(key)}>
+
+
+
+
+
+
+
+
+
+
+
+
+
+ {showMode === 'table' && (
+
+ )}
+ {showMode === 'chart' && (
+
+ {
+ return data.map((metricData: any) => {
+ const partitionMetricData = metricData.metricLine?.metricPoints || [];
+ return {
+ name: metricData.metricName,
+ data: partitionMetricData.map((item: any) => [item.timeStamp, item.value, item.unit]),
+ lineStyle: {
+ width: 1.5,
+ },
+ smooth: 0.25,
+ symbol: 'emptyCircle',
+ symbolSize: 4,
+ emphasis: {
+ disabled: true,
+ },
+ };
+ });
+ }}
+ />
+
+ )}
+
+
+
+
+
+
+
+
+
+
+
+
genData({ pageNo: pagination.current, pageSize: pagination.pageSize })}
+ >
+
+
+
+
+
+
+
+
+
+
+
+ {
}
+
Github:
- 5K
+ 5.4K
+ Star的项目 Know Streaming
diff --git a/km-console/packages/layout-clusters-fe/src/pages/Login/login.tsx b/km-console/packages/layout-clusters-fe/src/pages/Login/login.tsx
index 39c72a76..0400dbf0 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/Login/login.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/Login/login.tsx
@@ -1,5 +1,6 @@
import React from 'react';
-import { Form, Button, Input, Row, InputNumber, Utils, message } from 'knowdesign';
+import { Form, Button, Input, Row, InputNumber, Utils } from 'knowdesign';
+import message from '@src/components/Message';
import { FormMap } from './config';
import Api from '../../api';
import { useHistory } from 'react-router-dom';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
index 7435d9a7..e5c1f649 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
@@ -1,4 +1,5 @@
-import { Button, Divider, Drawer, Form, Input, InputNumber, message, Radio, Select, Spin, Space, Utils } from 'knowdesign';
+import { Button, Divider, Drawer, Form, Input, InputNumber, Radio, Select, Spin, Space, Utils } from 'knowdesign';
+import message from '@src/components/Message';
import * as React from 'react';
import { useIntl } from 'react-intl';
import api from '@src/api';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/List.tsx b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/List.tsx
index d15cffcb..18723cd8 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/List.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/List.tsx
@@ -1,4 +1,5 @@
-import { AppContainer, Divider, Form, Input, List, message, Modal, Progress, Spin, Tooltip, Utils } from 'knowdesign';
+import { AppContainer, Divider, Form, Input, List, Modal, Progress, Spin, Tooltip, Utils } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import moment from 'moment';
import API from '@src/api';
@@ -11,7 +12,7 @@ import { useIntl } from 'react-intl';
import api, { MetricType } from '@src/api';
import { getHealthClassName, getHealthProcessColor, getHealthText } from '../SingleClusterDetail/config';
import { ClustersPermissionMap } from '../CommonConfig';
-import { getUnit, getDataNumberUnit } from '@src/constants/chartConfig';
+import { getDataUnit } from '@src/constants/chartConfig';
import SmallChart from '@src/components/SmallChart';
import { SearchParams } from './HomePage';
@@ -235,14 +236,14 @@ const ClusterList = (props: { searchParams: SearchParams; showAccessCluster: any
// 如果单位是 字节 ，进行单位换算
if (line.unit.toLowerCase().includes('byte')) {
- const [unit, size] = getUnit(line.value);
+ const [unit, size] = getDataUnit['Memory'](line.value);
line.value = Number((line.value / size).toFixed(2));
line.unit = line.unit.toLowerCase().replace('byte', unit);
}
// Messages 指标值特殊处理
if (line.metricName === 'LeaderMessages') {
- const [unit, size] = getDataNumberUnit(line.value);
+ const [unit, size] = getDataUnit['Num'](line.value);
line.value = Number((line.value / size).toFixed(2));
line.unit = unit + line.unit;
}
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
index 9b2a7203..8afcdccc 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
@@ -1,5 +1,6 @@
import React, { forwardRef, useEffect, useImperativeHandle, useRef, useState } from 'react';
-import { Button, Form, Input, Select, message, Drawer, Space, Divider, Utils, Radio, AutoComplete, Alert } from 'knowdesign';
+import { Button, Form, Input, Select, Drawer, Space, Divider, Utils, Radio, AutoComplete, Alert } from 'knowdesign';
+import message from '@src/components/Message';
import api from '@src/api';
import { useParams } from 'react-router-dom';
import { UsersProps } from '../SecurityUsers';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
index a98f6269..d361f3e7 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
@@ -1,5 +1,6 @@
import React, { useEffect, useRef, useState } from 'react';
-import { Button, Form, Input, Select, Modal, message, ProTable, AppContainer, DKSBreadcrumb, Utils, Divider } from 'knowdesign';
+import { Button, Form, Input, Select, Modal, ProTable, AppContainer, DKSBreadcrumb, Utils, Divider } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import ACLsCardBar from '@src/components/CardBar/ACLsCardBar';
import api from '@src/api';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SecurityUsers/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/SecurityUsers/index.tsx
index 9f493bd7..6291ed42 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SecurityUsers/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SecurityUsers/index.tsx
@@ -4,7 +4,6 @@ import {
Form,
Input,
Modal,
- message,
ProTable,
Drawer,
Space,
@@ -16,6 +15,7 @@ import {
Tooltip,
Alert,
} from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import { CloseOutlined, EyeInvisibleOutlined, EyeOutlined, LoadingOutlined } from '@ant-design/icons';
import './index.less';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/config.tsx b/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/config.tsx
index 65ec2fd9..405f95c0 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/config.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/config.tsx
@@ -14,7 +14,7 @@ const messagesInTooltipFormatter = (date: any, arr: any) => {
-
+
${params.seriesName}
@@ -55,7 +55,7 @@ const messagesInTooltipFormatter = (date: any, arr: any) => {
};
export const getChartConfig = (props: any) => {
- const { metricName, lineColor, isDefaultMetric = false } = props;
+ const { lineColor, isDefaultMetric = false } = props;
return {
option: getBasicChartConfig({
// TODO: time 轴图表联动有问题，先切换为 category
@@ -63,7 +63,6 @@ export const getChartConfig = (props: any) => {
title: { show: false },
legend: { show: false },
grid: { top: 24, bottom: 12 },
- lineColor: [lineColor],
tooltip: isDefaultMetric
? {
formatter: function (params: any) {
@@ -87,6 +86,7 @@ export const getChartConfig = (props: any) => {
smooth: 0.25,
symbol: 'emptyCircle',
symbolSize: 4,
+ color: '#556ee6',
// 面积图样式
areaStyle: {
color: lineColor,
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/index.tsx
index 3df6ee1e..b9337f09 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SingleClusterDetail/DetailChart/index.tsx
@@ -5,16 +5,16 @@ import { arrayMoveImmutable } from 'array-move';
import api from '@src/api';
import { useParams } from 'react-router-dom';
import {
- MetricDefaultChartDataType,
- MetricChartDataType,
+ OriginMetricData,
+ FormattedMetricData,
formatChartData,
supplementaryPoints,
resolveMetricsRank,
MetricInfo,
} from '@src/constants/chartConfig';
import { MetricType } from '@src/api';
-import { getDataNumberUnit, getUnit } from '@src/constants/chartConfig';
-import SingleChartHeader, { KsHeaderOptions } from '@src/components/SingleChartHeader';
+import { getDataUnit } from '@src/constants/chartConfig';
+import ChartOperateBar, { KsHeaderOptions } from '@src/components/ChartOperateBar';
import RenderEmpty from '@src/components/RenderEmpty';
import DragGroup from '@src/components/DragGroup';
import { getChartConfig } from './config';
@@ -162,13 +162,13 @@ const DetailChart = (props: { children: JSX.Element }): JSX.Element => {
metricsNames: selectedMetricNames.filter((name) => name !== DEFAULT_METRIC),
},
}).then(
- (res: MetricDefaultChartDataType[]) => {
+ (res: OriginMetricData[]) => {
// 如果当前请求不是最新请求，则不做任何操作
if (curFetchingTimestamp.current.messagesIn !== curTimestamp) {
return;
}
- const formattedMetricData: MetricChartDataType[] = formatChartData(
+ const formattedMetricData: FormattedMetricData[] = formatChartData(
res,
global.getMetricDefine || {},
MetricType.Cluster,
@@ -224,7 +224,7 @@ const DetailChart = (props: { children: JSX.Element }): JSX.Element => {
let valueWithUnit = Number(value);
let unit = ((global.getMetricDefine && global.getMetricDefine(MetricType.Cluster, key)?.unit) || '') as string;
if (unit.toLowerCase().includes('byte')) {
- const [unitName, unitSize]: [string, number] = getUnit(Number(value));
+ const [unitName, unitSize]: [string, number] = getDataUnit['Memory'](Number(value));
unit = unit.toLowerCase().replace('byte', unitName);
valueWithUnit /= unitSize;
}
@@ -235,7 +235,7 @@ const DetailChart = (props: { children: JSX.Element }): JSX.Element => {
};
return returnValue;
});
- return [timeStamp, values.MessagesIn || '0', valuesWithUnit] as [number, number | string, typeof valuesWithUnit];
+ return [timeStamp, parsedValue || '0', valuesWithUnit] as [number, number | string, typeof valuesWithUnit];
});
result.sort((a, b) => (a[0] as number) - (b[0] as number));
const line = {
@@ -244,9 +244,9 @@ const DetailChart = (props: { children: JSX.Element }): JSX.Element => {
data: result as any,
};
if (maxValue > 0) {
- const [unitName, unitSize]: [string, number] = getDataNumberUnit(maxValue);
+ const [unitName, unitSize]: [string, number] = getDataUnit['Num'](maxValue);
line.unit = `${unitName}${line.unit}`;
- result.forEach((point) => ((point[1] as number) /= unitSize));
+ result.forEach((point) => parseFloat(((point[1] as number) /= unitSize).toFixed(3)));
}
if (result.length) {
@@ -308,11 +308,11 @@ const DetailChart = (props: { children: JSX.Element }): JSX.Element => {
return (
- {
item.configItem.indexOf('Group Re-Balance') > -1
? 'ReBalance'
: item.configItem.includes('副本未同步')
- ? 'UNDER_REPLICA'
- : item.configItem;
+ ? 'UNDER_REPLICA'
+ : item.configItem;
values[`weight_${item.configItemName}`] = itemValue?.weight;
values[`value_${item.configItemName}`] = itemValue?.value;
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TestingConsumer/Consume.tsx b/km-console/packages/layout-clusters-fe/src/pages/TestingConsumer/Consume.tsx
index be43d454..90625de8 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TestingConsumer/Consume.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TestingConsumer/Consume.tsx
@@ -1,6 +1,7 @@
/* eslint-disable no-case-declarations */
import { DownloadOutlined } from '@ant-design/icons';
-import { AppContainer, Divider, message, Tooltip, Utils } from 'knowdesign';
+import { AppContainer, Divider, Tooltip, Utils } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import * as React from 'react';
import moment from 'moment';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/Produce.tsx b/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/Produce.tsx
index 8f5e3479..18f0b2c5 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/Produce.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/Produce.tsx
@@ -1,4 +1,5 @@
-import { AppContainer, Form, message, Tabs, Utils } from 'knowdesign';
+import { AppContainer, Form, Tabs, Utils } from 'knowdesign';
+import message from '@src/components/Message';
import * as React from 'react';
import ConfigForm from './component/ConfigFrom';
import TestResult from '../TestingConsumer/component/Result';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/component/EditTable.tsx b/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/component/EditTable.tsx
index a5300af2..55b06e25 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/component/EditTable.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TestingProduce/component/EditTable.tsx
@@ -1,6 +1,7 @@
/* eslint-disable react/display-name */
import React, { useState } from 'react';
-import { Table, Input, InputNumber, Popconfirm, Form, Typography, Button, message, Select } from 'knowdesign';
+import { Table, Input, InputNumber, Popconfirm, Form, Typography, Button, Select } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import './style/edit-table.less';
import { CheckOutlined, CloseOutlined, PlusSquareOutlined } from '@ant-design/icons';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDashboard/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDashboard/index.tsx
index 72797227..ccf4104f 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDashboard/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDashboard/index.tsx
@@ -1,7 +1,7 @@
import React from 'react';
import { MetricType } from '@src/api';
import TopicHealthCheck from '@src/components/CardBar/TopicHealthCheck';
-import DashboardDragChart from '@src/components/DashboardDragChart';
+import DraggableCharts from '@src/components/DraggableCharts';
import { AppContainer } from 'knowdesign';
import DBreadcrumb from 'knowdesign/es/extend/d-breadcrumb';
@@ -19,7 +19,7 @@ const TopicDashboard = () => {
/>
-
+
);
};
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConfigurationEdit.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConfigurationEdit.tsx
index 7fc4ab80..ff8c6cd0 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConfigurationEdit.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConfigurationEdit.tsx
@@ -1,5 +1,6 @@
import React from 'react';
-import { Drawer, Form, Input, Space, Button, Utils, Row, Col, Divider, message } from 'knowdesign';
+import { Drawer, Form, Input, Space, Button, Utils, Row, Col, Divider } from 'knowdesign';
+import message from '@src/components/Message';
import { IconFont } from '@knowdesign/icons';
import { useParams } from 'react-router-dom';
import Api from '@src/api';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx
new file mode 100644
index 00000000..bd502b1a
--- /dev/null
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx
@@ -0,0 +1,502 @@
+import React, { useState, useEffect } from 'react';
+import { useParams, useHistory } from 'react-router-dom';
+import { AppContainer, Divider, Drawer, ProTable, Select, SingleChart, Space, Tooltip, Utils } from 'knowdesign';
+import { IconFont } from '@knowdesign/icons';
+import { DRangeTime } from 'knowdesign';
+import { CHART_COLOR_LIST, getBasicChartConfig } from '@src/constants/chartConfig';
+import Api from '@src/api/index';
+import { hashDataParse } from '@src/constants/common';
+import { ClustersPermissionMap } from '../CommonConfig';
+import ResetOffsetDrawer from './ResetOffsetDrawer';
+import SwitchTab from '@src/components/SwitchTab';
+import ContentWithCopy from '@src/components/CopyContent';
+
+const { Option } = Select;
+
+export interface MetricLine {
+ createTime?: number;
+ metricPoints: Array<{
+ aggType: string;
+ createTime: number;
+ timeStamp: number;
+ unit: string;
+ updateTime: number;
+ value: number;
+ }>;
+ name: string;
+ updateTime?: number;
+}
+export interface MetricData {
+ metricLines?: Array;
+ metricLine?: MetricLine;
+ metricName: string;
+}
+
+export interface HashData {
+ groupName: string;
+ topicName: string;
+}
+
+const metricConsts = ['LogEndOffset', 'OffsetConsumed', 'Lag'];
+const metricWithType = [
+ { metricName: 'LogEndOffset', metricType: 104 },
+ { metricName: 'OffsetConsumed', metricType: 102 },
+ { metricName: 'Lag', metricType: 102 },
+];
+
+export default (props: any) => {
+ const { scene, visible, setVisible, hashData } = props;
+ const params = useParams<{
+ clusterId: string;
+ }>();
+ const history = useHistory();
+ const [global] = AppContainer.useGlobalValue();
+ // const { record } = props;
+ const now = Date.now();
+ const [allGroupMetricsData, setAllGroupMetricsData] = useState>([]);
+ const [groupMetricsData, setGroupMetricsData] = useState>([]);
+ const [timeRange, setTimeRange] = useState([now - 24 * 60 * 60 * 1000, now]);
+ const [consumerList, setConsumerList] = useState([]);
+ const [partitionList, setPartitionList] = useState([]);
+ const [curPartition, setCurPartition] = useState('');
+ const [showMode, setShowMode] = useState('table');
+ const [pageIndex, setPageIndex] = useState(1);
+ const [pageTotal, setPageTotal] = useState(1);
+ const [pageSize, setPageSize] = useState(10);
+ const [consumerListLoading, setConsumerListLoading] = useState(true);
+ const [consumerChartLoading, setConsumerChartLoading] = useState(false);
+ // const [hashData, setHashData] = useState({ groupName: '', topicName: '' });
+ // const [visible, setVisible] = useState(false);
+ const [sortObj, setSortObj] = useState<{
+ sortField: string;
+ sortType: 'desc' | 'asc' | '';
+ }>({ sortField: '', sortType: '' });
+ const [pagination, setPagination] = useState({
+ current: 1,
+ pageSize: 10,
+ position: 'bottomRight',
+ showSizeChanger: true,
+ pageSizeOptions: ['10', '20', '50', '100', '200', '500'],
+ showTotal: (total: number) => `共 ${total} 条目`,
+ });
+ const clusterId = Number(params.clusterId);
+ const columns = [
+ {
+ title: 'Topic Partition',
+ dataIndex: 'partitionId',
+ key: 'partitionId',
+ lineClampOne: true,
+ needTooltip: true,
+ width: 180,
+ render: (v: string, record: any) => {
+ return `${record.topicName}-${v}`;
+ },
+ },
+ {
+ title: 'Member ID',
+ dataIndex: 'memberId',
+ key: 'memberId',
+ width: 200,
+ render: (v: string) => {
+ return v ? : '-';
+ },
+ },
+ {
+ title: 'Current Offset',
+ dataIndex: 'OffsetConsumed',
+ key: 'OffsetConsumed',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.OffsetConsumed.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'OffsetConsumed' && item.metricType === 102)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'OffsetConsumed' && item.metricType === 102)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 1
+ // }
+ },
+ {
+ title: 'Log End Offset',
+ dataIndex: 'LogEndOffset',
+ key: 'LogEndOffset',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.LogEndOffset.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'LogEndOffset' && item.metricType === 104)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'LogEndOffset' && item.metricType === 104)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 2
+ // }
+ },
+ {
+ title: 'Lag',
+ dataIndex: 'Lag',
+ key: 'Lag',
+ render: (v: any, record: any) => {
+ return record?.latestMetrics?.metrics?.Lag.toLocaleString();
+ },
+ sorter: true,
+ // sorter: {
+ // compare: (a: any, b: any) => {
+ // let value1 = a?.metrics?.find((item: any) => item.metricName === 'Lag' && item.metricType === 102)?.metricValue
+ // let value2 = b?.metrics?.find((item: any) => item.metricName === 'Lag' && item.metricType === 102)?.metricValue
+ // return value1 - value2
+ // },
+ // multiple: 3
+ // }
+ },
+ {
+ title: 'Host',
+ dataIndex: 'host',
+ key: 'host',
+ },
+ {
+ title: 'Client ID',
+ dataIndex: 'clientId',
+ key: 'clientId',
+ needTooltip: true,
+ lineClampOne: true,
+ width: 200,
+ },
+ ];
+ const getTopicGroupMetric = ({
+ hashData,
+ pagination = { current: 1, pageSize: 10 },
+ sorter = {},
+ }: {
+ hashData: HashData;
+ pagination?: any;
+ sorter?: any;
+ }) => {
+ setConsumerListLoading(true);
+ const params: any = {
+ // metricRealTimes: metricWithType,
+ latestMetricNames: metricConsts,
+ pageNo: pagination.current,
+ pageSize: pagination.pageSize,
+ sortField: sorter.field || undefined,
+ sortType: sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : undefined,
+ };
+ // if (sorter.sortField && sorter.sortType) {
+ // params.sortField = sorter.sortField;
+ // params.sortType = sorter.sortType;
+ // }
+ return Utils.post(
+ Api.getTopicGroupMetric({
+ clusterId,
+ groupName: hashData.groupName,
+ topicName: hashData.topicName,
+ }),
+ params
+ )
+ .then((data: any) => {
+ if (!data) return;
+
+ setPagination({
+ current: data.pagination?.pageNo,
+ pageSize: data.pagination?.pageSize,
+ total: data.pagination?.total,
+ });
+ setConsumerList(data?.bizData);
+ })
+ .finally(() => {
+ setConsumerListLoading(false);
+ });
+ };
+ const getTopicGroupPartitionsHistory = (hashData: HashData) => {
+ return Utils.request(Api.getTopicGroupPartitionsHistory(clusterId, hashData.groupName), {
+ params: {
+ startTime: timeRange[0],
+ endTime: timeRange[1],
+ },
+ });
+ };
+ const getTopicGroupMetricHistory = (partitions: Array, hashData: HashData) => {
+ const params = {
+ aggType: 'sum',
+ groupTopics: partitions?.map((p) => ({
+ partition: p.value,
+ topic: hashData.topicName,
+ })),
+ group: hashData.groupName,
+ metricsNames: metricWithType.map((item) => item.metricName),
+ startTime: timeRange[0],
+ endTime: timeRange[1],
+ topNu: 0,
+ };
+ Utils.post(Api.getTopicGroupMetricHistory(clusterId), params).then((data: Array) => {
+ setAllGroupMetricsData(data);
+ });
+ };
+ const getConsumersMetadata = (hashData: HashData) => {
+ return Utils.request(Api.getConsumersMetadata(clusterId, hashData.groupName, hashData.topicName));
+ };
+
+ const getTopicsMetaData = (hashData: HashData) => {
+ return Utils.request(Api.getTopicsMetaData(hashData.topicName, clusterId));
+ };
+
+ const onClose = () => {
+ setVisible(false);
+ setSortObj({
+ sortField: '',
+ sortType: '',
+ });
+ // clean hash'
+ // scene === 'topicDetail' && history.goBack();
+ // scene !== 'topicDetail' && window.history.pushState('', '', location.pathname);
+ };
+
+ const onTableChange = (pagination: any, filters: any, sorter: any) => {
+ getTopicGroupMetric({ hashData, pagination, sorter });
+ // setPageIndex(pagination.current);
+ };
+
+ useEffect(() => {
+ if (curPartition === '' || allGroupMetricsData.length === 0) return;
+ const filteredData = allGroupMetricsData.map((item) => {
+ const allData = item.metricLines.reduce(
+ (acc, cur) => {
+ if (acc.metricLine.metricPoints.length === 0) {
+ acc.metricLine.metricPoints = cur.metricPoints.map((p) => ({
+ timeStamp: p.timeStamp,
+ value: Number(p.value),
+ }));
+ } else {
+ acc.metricLine.metricPoints.forEach((mp) => {
+ const curMetricPoint = cur.metricPoints.find((curmp) => curmp.timeStamp === mp.timeStamp);
+ mp.value += curMetricPoint ? Number(curMetricPoint.value) : 0;
+ });
+ }
+ return acc;
+ },
+ {
+ metricName: item.metricName,
+ metricLine: {
+ name: 'all',
+ metricPoints: [],
+ },
+ }
+ );
+ return curPartition === '__all__'
+ ? allData
+ : {
+ metricName: item.metricName,
+ metricLine: item.metricLines.find((line) => line.name.indexOf(curPartition) >= 0),
+ };
+ });
+ setGroupMetricsData(filteredData);
+ }, [curPartition, allGroupMetricsData]);
+
+ useEffect(() => {
+ // const hashData = hashDataParse(location.hash);
+ if (!hashData.groupName || !hashData.topicName) return;
+ // setHashData(hashData);
+ // 获取分区列表 为图表模式做准备
+ visible &&
+ getConsumersMetadata(hashData).then((res: any) => {
+ if (!res.exist) {
+ setVisible(false);
+ // history.push(`/cluster/${params?.clusterId}/consumers`);
+ return;
+ }
+ getTopicsMetaData(hashData)
+ // .then((data: any) => {
+ // if (data.length > 0) {
+ // setCurPartition(data[0].partition);
+ // }
+ // setPartitionList(data);
+ // return data;
+ // })
+ .then((data: any) => {
+ const partitionLists = (data?.partitionIdList || []).map((item: any) => {
+ return {
+ label: item,
+ value: item,
+ };
+ });
+ setCurPartition(partitionLists?.[0]?.value);
+ setPartitionList(partitionLists);
+ getTopicGroupMetricHistory(partitionLists, hashData);
+ })
+ .catch((e) => {
+ // history.push(`/cluster/${params?.clusterId}/consumers`);
+ setVisible(false);
+ });
+ // 获取Consumer列表 表格模式
+ getTopicGroupMetric({ hashData: hashData as HashData });
+ });
+ }, [visible]);
+
+ useEffect(() => {
+ if (partitionList.length === 0) return;
+ getTopicGroupMetricHistory(partitionList, hashData);
+ }, [timeRange]);
+
+ return (
+
+ {global.hasPermission &&
+ global.hasPermission(
+ scene === 'topicDetail' ? ClustersPermissionMap.TOPIC_RESET_OFFSET : ClustersPermissionMap.CONSUMERS_RESET_OFFSET
+ ) && }
+
+
+ }
+ >
+
+
+
+
+ {showMode === 'chart' && (
+
+ )}
+ {showMode === 'chart' && (
+
{
+ setTimeRange(o);
+ }}
+ >
+ )}
+ {showMode === 'chart' &&
}
+
setShowMode(key)}>
+
+
+
+
+
+
+
+
+
+
+
+
+
+ {showMode === 'table' && (
+ //
{
+ // setSortObj({
+ // sortField: sorter.field || '',
+ // sortType: sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : '',
+ // });
+ // setPageIndex(pagination.current);
+ // }}
+ // >
+
+ )}
+ {showMode === 'chart' && (
+
+ {
+ return data.map((metricData: any) => {
+ const partitionMetricData = metricData.metricLine?.metricPoints || [];
+ return {
+ name: metricData.metricName,
+ data: partitionMetricData.map((item: any) => [item.timeStamp, item.value, item.unit]),
+ lineStyle: {
+ width: 1.5,
+ },
+ smooth: 0.25,
+ symbol: 'emptyCircle',
+ symbolSize: 4,
+ emphasis: {
+ disabled: true,
+ },
+ };
+ });
+ }}
+ />
+
+ )}
+
Topic名称 :
- {record.topicName}
+
+ {record.topicName}
+
描述 :
- {record.description}
+
+ {record.description || '-'}
+
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.less b/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.less
index 453363c6..0fa44077 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.less
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.less
@@ -142,6 +142,7 @@
color: #74788d;
}
.desc-field {
+ flex-shrink: 0;
width: 34px;
}
.val,
@@ -149,17 +150,28 @@
color: #495057;
letter-spacing: 0;
font-weight: 400;
- }
- .val {
- width: 105px;
margin-left: 12px;
}
- .desc-val {
- width: 809px;
- height: 36px;
+ .val {
+ max-width: 208px;
overflow: hidden;
- -webkit-box-orient: vertical;
+ text-overflow: ellipsis;
+ white-space: nowrap;
+ }
+ .desc-val {
+ height: 36px;
+ display: -webkit-box;
+ overflow: hidden;
+ text-overflow: ellipsis;
-webkit-line-clamp: 2;
+ -webkit-box-orient: vertical;
+ word-break: break-all;
+ }
+ &:first-child {
+ margin-right: 40px;
+ }
+ &:last-child {
+ flex: 1;
}
}
}
diff --git a/km-console/packages/layout-clusters-fe/src/pages/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/index.tsx
index 4e1d1808..16d73de9 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/index.tsx
@@ -1,6 +1,7 @@
import React, { useState, useEffect, useCallback } from 'react';
import { Redirect, useHistory, useLocation } from 'react-router-dom';
-import { DProLayout, AppContainer, RouteGuard, notification, Spin } from 'knowdesign';
+import { DProLayout, AppContainer, RouteGuard, Spin } from 'knowdesign';
+import notification from '@src/components/Notification';
import { pageRoutes } from './pageRoutes';
import { leftMenus, systemKey } from '@src/constants/menu';
import { ClustersPermissionMap } from './CommonConfig';
diff --git a/km-console/packages/layout-clusters-fe/src/pages/pageRoutes.ts b/km-console/packages/layout-clusters-fe/src/pages/pageRoutes.ts
index 13422a46..50e6ac71 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/pageRoutes.ts
+++ b/km-console/packages/layout-clusters-fe/src/pages/pageRoutes.ts
@@ -10,7 +10,7 @@ import BrokerControllerChangeLog from './BrokerControllerChangeLog';
import TopicBoard from './TopicDashboard';
import TopicList from './TopicList';
-import Consumers from './Consumers/index';
+import Consumers from './ConsumerGroup';
import Jobs from './Jobs';
diff --git a/km-console/packages/layout-clusters-fe/src/style-addition.less b/km-console/packages/layout-clusters-fe/src/style-addition.less
index 451c31bf..f4b7e04b 100644
--- a/km-console/packages/layout-clusters-fe/src/style-addition.less
+++ b/km-console/packages/layout-clusters-fe/src/style-addition.less
@@ -11,6 +11,7 @@
@select-item-selected-font-weight: normal;
@btn-danger-bg: #F5483B;
@btn-danger-border: #F5483B;
+@notification-icon-size: 20px;
// 自定义变量
// Input
@input-bg: rgba(33, 37, 41, 0.06);
@@ -656,3 +657,109 @@
.@{ant-prefix}-empty-img-default{
width: 100% !important;
}
+
+// message 样式覆盖
+.@{message-prefix-cls} {
+ &-notice-content {
+ padding: 8px 12px;
+ font-size: 14px;
+ border-radius: 4px;
+ background: #fff;
+ }
+ &-success,
+ &-info,
+ &-warning,
+ &-error,
+ &-loading {
+ margin: unset;
+ padding: unset;
+ color: #000;
+ border-style: unset;
+ border-width: unset;
+ border-radius: unset;
+ }
+
+ &-success {
+ background-color: unset;
+ border-color: unset;
+ }
+
+ &-info,
+ &-loading {
+ background-color: unset;
+ border-color: unset;
+ }
+
+ &-warning {
+ background-color: unset;
+ border-color: unset;
+ }
+
+ &-error {
+ background-color: unset;
+ border-color: unset;
+ }
+
+ .@{iconfont-css-prefix} {
+ top: 2px;
+ color: unset;
+ }
+}
+
+// Notification 样式覆盖
+.@{notification-prefix-cls} {
+ &-notice {
+ width: 348px;
+ color: #000;
+ background-color: #fff;
+ border: unset;
+ border-radius: 4px;
+ box-shadow: 0 2px 4px 0 rgba(0,0,0,0.02), 0 4px 6px 6px rgba(0,0,0,0.02), 0 4px 6px 0 rgba(0,0,0,0.06);
+ &-message {
+ font-family: @font-family-bold;
+ font-size: 16px;
+ color: #000;
+ }
+ &-description {
+ font-size: 14px;
+ color: rgba(0,0,0,0.60);
+ }
+ &-with-icon &-message {
+ font-size: 16px;
+ margin-left: 26px;
+ }
+ &-with-icon &-description {
+ font-size: 14px;
+ margin-left: 26px;
+ }
+ &-icon {
+ top: 18px;
+ font-size: @notification-icon-size;
+ line-height: @notification-icon-size;
+ }
+ .@{iconfont-css-prefix}&-icon {
+ color: @notification-content-color !important;
+ }
+ &-success {
+ background-color: #fff;
+ border: unset;
+ }
+ &-warning {
+ background-color: #fff;
+ border: unset;
+ }
+ &-error {
+ background-color: #fff;
+ border: unset;
+ }
+ &-close {
+ color: rgba(0,0,0,0.60);
+
+ &:hover {
+ & when not (@theme = dark) {
+ color: shade(@white, 30%);
+ }
+ }
+ }
+ }
+}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/CollectedMetricsLocalCache.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/CollectedMetricsLocalCache.java
index bc5b1c34..2fc0a4ff 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/CollectedMetricsLocalCache.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/CollectedMetricsLocalCache.java
@@ -24,11 +24,6 @@ public class CollectedMetricsLocalCache {
.maximumSize(10000)
.build();
- private static final Cache replicaMetricsValueCache = Caffeine.newBuilder()
- .expireAfterWrite(90, TimeUnit.SECONDS)
- .maximumSize(20000)
- .build();
-
public static Float getBrokerMetrics(String brokerMetricKey) {
return brokerMetricsCache.getIfPresent(brokerMetricKey);
}
@@ -64,17 +59,6 @@ public class CollectedMetricsLocalCache {
partitionMetricsCache.put(partitionMetricsKey, metricsList);
}
- public static Float getReplicaMetrics(String replicaMetricsKey) {
- return replicaMetricsValueCache.getIfPresent(replicaMetricsKey);
- }
-
- public static void putReplicaMetrics(String replicaMetricsKey, Float value) {
- if (value == null) {
- return;
- }
- replicaMetricsValueCache.put(replicaMetricsKey, value);
- }
-
public static String genBrokerMetricKey(Long clusterPhyId, Integer brokerId, String metricName) {
return clusterPhyId + "@" + brokerId + "@" + metricName;
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/ZookeeperLocalCache.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/ZookeeperLocalCache.java
new file mode 100644
index 00000000..24f6c7f2
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/ZookeeperLocalCache.java
@@ -0,0 +1,46 @@
+package com.xiaojukeji.know.streaming.km.core.cache;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.BaseFourLetterWordCmdData;
+
+import java.util.concurrent.TimeUnit;
+
+public class ZookeeperLocalCache {
+ private static final Cache fourLetterCmdFailedServerCache = Caffeine.newBuilder()
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .maximumSize(10000)
+ .build();
+
+ private static final Cache fourLetterCmdDataCache = Caffeine.newBuilder()
+ .expireAfterWrite(60, TimeUnit.SECONDS)
+ .maximumSize(10000)
+ .build();
+
+ public static boolean canUse(String host, int port, String cmd) {
+ String data = fourLetterCmdFailedServerCache.getIfPresent(gen4lwFailedKey(host, port, cmd));
+
+ return data == null;
+ }
+
+ public static void setFailed(String host, int port, String cmd) {
+ fourLetterCmdFailedServerCache.put(gen4lwFailedKey(host, port, cmd), "");
+ }
+
+ public static BaseFourLetterWordCmdData getData(String host, int port, String cmd) {
+ return fourLetterCmdDataCache.getIfPresent(gen4lwFailedKey(host, port, cmd));
+ }
+
+ public static void putData(String host, int port, String cmd, BaseFourLetterWordCmdData cmdData) {
+ fourLetterCmdDataCache.put(gen4lwFailedKey(host, port, cmd), cmdData);
+ }
+
+ /**************************************************** private method ****************************************************/
+
+ private static String gen4lwFailedKey(String host, int port, String cmd) {
+ return host + "@" + port + "@" + cmd;
+ }
+
+
+
+}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java
index 62f03e65..c8c300a0 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java
@@ -67,4 +67,8 @@ public interface BrokerService {
* 获取总的Broker数
*/
Integer countAllBrokers();
+
+ boolean allServerDown(Long clusterPhyId);
+
+ boolean existServerDown(Long clusterPhyId);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerMetricServiceImpl.java
index 93c343ff..e82882e1 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerMetricServiceImpl.java
@@ -37,6 +37,8 @@ import com.xiaojukeji.know.streaming.km.core.service.version.metrics.BrokerMetri
import com.xiaojukeji.know.streaming.km.core.service.version.metrics.ReplicaMetricVersionItems;
import com.xiaojukeji.know.streaming.km.persistence.es.dao.BrokerMetricESDAO;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaJMXClient;
+import org.apache.kafka.clients.admin.LogDirDescription;
+import org.apache.kafka.clients.admin.ReplicaInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@@ -49,6 +51,7 @@ import java.util.*;
import java.util.stream.Collectors;
import static com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus.*;
+import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionEnum.*;
/**
* @author didi
@@ -105,7 +108,11 @@ public class BrokerMetricServiceImpl extends BaseMetricService implements Broker
registerVCHandler( BROKER_METHOD_GET_HEALTH_SCORE, this::getMetricHealthScore);
registerVCHandler( BROKER_METHOD_GET_PARTITIONS_SKEW, this::getPartitionsSkew);
registerVCHandler( BROKER_METHOD_GET_LEADERS_SKEW, this::getLeadersSkew);
- registerVCHandler( BROKER_METHOD_GET_LOG_SIZE, this::getLogSize);
+// registerVCHandler( BROKER_METHOD_GET_LOG_SIZE, this::getLogSize);
+
+ registerVCHandler( BROKER_METHOD_GET_LOG_SIZE, V_0_10_0_0, V_1_0_0, "getLogSizeFromJmx", this::getLogSizeFromJmx);
+ registerVCHandler( BROKER_METHOD_GET_LOG_SIZE, V_1_0_0, V_MAX, "getLogSizeFromClient", this::getLogSizeFromClient);
+
registerVCHandler( BROKER_METHOD_IS_BROKER_ALIVE, this::isBrokerAlive);
}
@@ -351,7 +358,7 @@ public class BrokerMetricServiceImpl extends BaseMetricService implements Broker
);
}
- private Result getLogSize(VersionItemParam metricParam) {
+ private Result getLogSizeFromJmx(VersionItemParam metricParam) {
BrokerMetricParam param = (BrokerMetricParam)metricParam;
String metric = param.getMetric();
@@ -360,19 +367,17 @@ public class BrokerMetricServiceImpl extends BaseMetricService implements Broker
List partitions = partitionService.listPartitionByBroker(clusterId, brokerId);
- JmxConnectorWrap jmxConnectorWrap = kafkaJMXClient.getClientWithCheck(clusterId, brokerId);
- if (ValidateUtils.isNull(jmxConnectorWrap)){return Result.buildFailure(VC_JMX_INIT_ERROR);}
-
Float logSizeSum = 0f;
for(Partition p : partitions) {
try {
- Result metricsResult = replicaMetricService.collectReplicaMetricsFromKafkaWithCache(
+ Result metricsResult = replicaMetricService.collectReplicaMetricsFromKafka(
clusterId,
p.getTopicName(),
brokerId,
p.getPartitionId(),
ReplicaMetricVersionItems.REPLICATION_METRIC_LOG_SIZE
);
+
if(null == metricsResult || metricsResult.failed() || null == metricsResult.getData()) {
continue;
}
@@ -391,6 +396,28 @@ public class BrokerMetricServiceImpl extends BaseMetricService implements Broker
return Result.buildSuc(BrokerMetrics.initWithMetric(clusterId, brokerId, metric, logSizeSum));
}
+ private Result getLogSizeFromClient(VersionItemParam metricParam) {
+ BrokerMetricParam param = (BrokerMetricParam)metricParam;
+
+ String metric = param.getMetric();
+ Long clusterId = param.getClusterId();
+ Integer brokerId = param.getBrokerId();
+
+ Result> descriptionMapResult = brokerService.getBrokerLogDirDescFromKafka(clusterId, brokerId);
+ if(null == descriptionMapResult || descriptionMapResult.failed() || null == descriptionMapResult.getData()) {
+ return Result.buildFromIgnoreData(descriptionMapResult);
+ }
+
+ Float logSizeSum = 0f;
+ for (LogDirDescription logDirDescription: descriptionMapResult.getData().values()) {
+ for (ReplicaInfo replicaInfo: logDirDescription.replicaInfos().values()) {
+ logSizeSum += replicaInfo.size();
+ }
+ }
+
+ return Result.buildSuc(BrokerMetrics.initWithMetric(clusterId, brokerId, metric, logSizeSum));
+ }
+
private Result getLeadersSkew(VersionItemParam metricParam) {
BrokerMetricParam param = (BrokerMetricParam)metricParam;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java
index fbede23c..3fd74ee5 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java
@@ -262,14 +262,32 @@ public class BrokerServiceImpl extends BaseVersionControlService implements Brok
return version;
}
-
-
@Override
public Integer countAllBrokers() {
LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
return brokerDAO.selectCount(lambdaQueryWrapper);
}
+ @Override
+ public boolean allServerDown(Long clusterPhyId) {
+ List poList = this.getAllBrokerPOsFromDB(clusterPhyId);
+ if (ValidateUtils.isEmptyList(poList)) {
+ return false;
+ }
+
+ return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() == poList.size();
+ }
+
+ @Override
+ public boolean existServerDown(Long clusterPhyId) {
+ List poList = this.getAllBrokerPOsFromDB(clusterPhyId);
+ if (ValidateUtils.isEmptyList(poList)) {
+ return false;
+ }
+
+ return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() > 0;
+ }
+
/**************************************************** private method ****************************************************/
private List listAllBrokersAndUpdateCache(Long clusterPhyId) {
@@ -343,17 +361,9 @@ public class BrokerServiceImpl extends BaseVersionControlService implements Brok
private Broker getStartTimeAndBuildBroker(Long clusterPhyId, Node newNode, JmxConfig jmxConfig) {
try {
- Object object = jmxDAO.getJmxValue(
- clusterPhyId,
- newNode.id(),
- newNode.host(),
- null,
- jmxConfig,
- new ObjectName("java.lang:type=Runtime"),
- "StartTime"
- );
+ Long startTime = jmxDAO.getServerStartTime(clusterPhyId, newNode.host(), null, jmxConfig);
- return Broker.buildFrom(clusterPhyId, newNode, object != null? (Long) object: null);
+ return Broker.buildFrom(clusterPhyId, newNode, startTime);
} catch (Exception e) {
log.error("class=BrokerServiceImpl||method=getStartTimeAndBuildBroker||clusterPhyId={}||brokerNode={}||jmxConfig={}||errMsg=exception!", clusterPhyId, newNode, jmxConfig, e);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterMetricServiceImpl.java
index 9fdd9ec0..3d004f78 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterMetricServiceImpl.java
@@ -85,7 +85,7 @@ public class ClusterMetricServiceImpl extends BaseMetricService implements Clust
public static final String CLUSTER_METHOD_GET_TOTAL_LOG_SIZE = "getTotalLogSize";
public static final String CLUSTER_METHOD_GET_PARTITION_SIZE = "getPartitionSize";
public static final String CLUSTER_METHOD_GET_PARTITION_NO_LEADER_SIZE = "getPartitionNoLeaderSize";
- public static final String CLUSTER_METHOD_GET_HEALTH_SCORE = "getTopicHealthScore";
+ public static final String CLUSTER_METHOD_GET_HEALTH_SCORE = "getClusterHealthScore";
public static final String CLUSTER_METHOD_GET_METRIC_FROM_KAFKA_BY_TOTAL_BROKERS_JMX = "getMetricFromKafkaByTotalBrokersJMX";
public static final String CLUSTER_METHOD_GET_METRIC_FROM_KAFKA_BY_CONTROLLER_JMX = "getMetricFromKafkaByControllerJMX";
public static final String CLUSTER_METHOD_GET_ZK_COUNT = "getZKCount";
@@ -188,7 +188,7 @@ public class ClusterMetricServiceImpl extends BaseMetricService implements Clust
registerVCHandler( CLUSTER_METHOD_GET_PARTITION_SIZE, this::getPartitionSize);
registerVCHandler( CLUSTER_METHOD_GET_PARTITION_NO_LEADER_SIZE, this::getPartitionNoLeaderSize);
- registerVCHandler( CLUSTER_METHOD_GET_HEALTH_SCORE, this::getTopicHealthScore);
+ registerVCHandler( CLUSTER_METHOD_GET_HEALTH_SCORE, this::getClusterHealthScore);
registerVCHandler( CLUSTER_METHOD_GET_METRIC_FROM_KAFKA_BY_TOTAL_BROKERS_JMX, this::getMetricFromKafkaByTotalBrokersJMX);
registerVCHandler( CLUSTER_METHOD_GET_METRIC_FROM_KAFKA_BY_CONTROLLER_JMX, this::getMetricFromKafkaByControllerJMX);
@@ -364,7 +364,7 @@ public class ClusterMetricServiceImpl extends BaseMetricService implements Clust
/**
* 获取集群的健康分
*/
- private Result getTopicHealthScore(VersionItemParam metricParam){
+ private Result getClusterHealthScore(VersionItemParam metricParam){
ClusterMetricParam param = (ClusterMetricParam)metricParam;
ClusterMetrics clusterMetrics = healthScoreService.calClusterHealthScore(param.getClusterId());
return Result.buildSuc(clusterMetrics);
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/GroupService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/GroupService.java
index 790a7c47..8dc1c535 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/GroupService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/GroupService.java
@@ -1,6 +1,7 @@
package com.xiaojukeji.know.streaming.km.core.service.group;
import com.xiaojukeji.know.streaming.km.common.bean.dto.pagination.PaginationBaseDTO;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.group.Group;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.PaginationResult;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.po.group.GroupMemberPO;
@@ -16,27 +17,47 @@ import java.util.Map;
public interface GroupService {
/**
- * 从Kafka中获取消费组
- * @param clusterPhyId 集群ID
- * @return
- * @throws NotExistException
- * @throws AdminOperateException
+ * 从Kafka中获取消费组名称列表
*/
List listGroupsFromKafka(Long clusterPhyId) throws NotExistException, AdminOperateException;
- Map getGroupOffset(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException;
+ /**
+ * 从Kafka中获取消费组详细信息
+ */
+ Group getGroupFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException;
- ConsumerGroupDescription getGroupDescription(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException;
+ Map getGroupOffsetFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException;
- int replaceDBData(GroupMemberPO groupMemberPO);
+ ConsumerGroupDescription getGroupDescriptionFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException;
- void batchReplace(List newGroupMemberList);
+ Result resetGroupOffsets(Long clusterPhyId, String groupName, Map offsetMap, String operator) throws NotExistException, AdminOperateException;
+ /**
+ * 批量更新DB
+ */
+ void batchReplaceGroupsAndMembers(Long clusterPhyId, List newGroupList, long updateTime);
+
+ int deleteByUpdateTimeBeforeInDB(Long clusterPhyId, Date beforeTime);
+
+ /**
+ * DB-Group相关接口
+ */
GroupStateEnum getGroupStateFromDB(Long clusterPhyId, String groupName);
- List listGroupByTopic(Long clusterPhyId, String topicName);
+ Group getGroupFromDB(Long clusterPhyId, String groupName);
- List listGroup(Long clusterPhyId);
+ List listClusterGroups(Long clusterPhyId);
+
+ List getGroupsFromDB(Long clusterPhyId);
+
+ Integer calGroupCount(Long clusterPhyId);
+
+ Integer calGroupStatCount(Long clusterPhyId, GroupStateEnum stateEnum);
+
+ /**
+ * DB-GroupTopic相关接口
+ */
+ List listGroupByTopic(Long clusterPhyId, String topicName);
PaginationResult pagingGroupMembers(Long clusterPhyId,
String topicName,
@@ -45,15 +66,5 @@ public interface GroupService {
String searchGroupKeyword,
PaginationBaseDTO dto);
- int deleteByUpdateTimeBeforeInDB(Long clusterPhyId, Date beforeTime);
-
- List getGroupsFromDB(Long clusterPhyId);
-
- GroupMemberPO getGroupFromDB(Long clusterPhyId, String groupName, String topicName);
-
- Integer calGroupCount(Long clusterPhyId);
-
- Integer calGroupStatCount(Long clusterPhyId, GroupStateEnum stateEnum);
-
- Result resetGroupOffsets(Long clusterPhyId, String groupName, Map offsetMap, String operator) throws NotExistException, AdminOperateException;
-}
+ GroupMemberPO getGroupTopicFromDB(Long clusterPhyId, String groupName, String topicName);
+}
\ No newline at end of file
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
index 427edc2c..936897a3 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
@@ -24,7 +24,6 @@ import com.xiaojukeji.know.streaming.km.core.service.health.score.HealthScoreSer
import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService;
import com.xiaojukeji.know.streaming.km.core.service.version.BaseMetricService;
import com.xiaojukeji.know.streaming.km.persistence.es.dao.GroupMetricESDAO;
-import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
@@ -183,7 +182,7 @@ public class GroupMetricServiceImpl extends BaseMetricService implements GroupMe
List metricsList = new ArrayList<>();
try {
- Map groupOffsetMap = groupService.getGroupOffset(clusterId, groupName);
+ Map groupOffsetMap = groupService.getGroupOffsetFromKafka(clusterId, groupName);
// 组织 GROUP_METRIC_OFFSET_CONSUMED 指标
for (Map.Entry entry: groupOffsetMap.entrySet()) {
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
index 4cf29d2a..1a923f21 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
@@ -7,11 +7,15 @@ import com.didiglobal.logi.log.ILog;
import com.didiglobal.logi.log.LogFactory;
import com.didiglobal.logi.security.common.dto.oplog.OplogDTO;
import com.xiaojukeji.know.streaming.km.common.bean.dto.pagination.PaginationBaseDTO;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.group.Group;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.group.GroupTopicMember;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.PaginationResult;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
import com.xiaojukeji.know.streaming.km.common.bean.po.group.GroupMemberPO;
+import com.xiaojukeji.know.streaming.km.common.bean.po.group.GroupPO;
import com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant;
+import com.xiaojukeji.know.streaming.km.common.converter.GroupConverter;
import com.xiaojukeji.know.streaming.km.common.enums.group.GroupStateEnum;
import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.ModuleEnum;
import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.OperationEnum;
@@ -24,6 +28,7 @@ import com.xiaojukeji.know.streaming.km.core.service.group.GroupService;
import com.xiaojukeji.know.streaming.km.core.service.oprecord.OpLogWrapService;
import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
+import com.xiaojukeji.know.streaming.km.persistence.mysql.group.GroupDAO;
import com.xiaojukeji.know.streaming.km.persistence.mysql.group.GroupMemberDAO;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
@@ -41,6 +46,9 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
public class GroupServiceImpl extends BaseVersionControlService implements GroupService {
private static final ILog log = LogFactory.getLog(GroupServiceImpl.class);
+ @Autowired
+ private GroupDAO groupDAO;
+
@Autowired
private GroupMemberDAO groupMemberDAO;
@@ -79,7 +87,43 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
}
@Override
- public Map getGroupOffset(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException {
+ public Group getGroupFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException {
+ // 获取消费组的详细信息
+ ConsumerGroupDescription groupDescription = this.getGroupDescriptionFromKafka(clusterPhyId, groupName);
+ if (groupDescription == null) {
+ return null;
+ }
+
+ Group group = new Group(clusterPhyId, groupName, groupDescription);
+
+ // 获取消费组消费过哪些Topic
+ Map memberMap = new HashMap<>();
+ for (TopicPartition tp : this.getGroupOffsetFromKafka(clusterPhyId, groupName).keySet()) {
+ memberMap.putIfAbsent(tp.topic(), new GroupTopicMember(tp.topic(), 0));
+ }
+
+ // 记录成员信息
+ for (MemberDescription memberDescription : groupDescription.members()) {
+ Set partitionList = new HashSet<>();
+ if (!ValidateUtils.isNull(memberDescription.assignment().topicPartitions())) {
+ partitionList = memberDescription.assignment().topicPartitions();
+ }
+
+ Set topicNameSet = partitionList.stream().map(elem -> elem.topic()).collect(Collectors.toSet());
+ for (String topicName : topicNameSet) {
+ memberMap.putIfAbsent(topicName, new GroupTopicMember(topicName, 0));
+
+ GroupTopicMember member = memberMap.get(topicName);
+ member.setMemberCount(member.getMemberCount() + 1);
+ }
+ }
+ group.setTopicMembers(memberMap.values().stream().collect(Collectors.toList()));
+
+ return group;
+ }
+
+ @Override
+ public Map getGroupOffsetFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException {
AdminClient adminClient = kafkaAdminClient.getClient(clusterPhyId);
Map offsetMap = new HashMap<>();
@@ -99,7 +143,7 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
}
@Override
- public ConsumerGroupDescription getGroupDescription(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException {
+ public ConsumerGroupDescription getGroupDescriptionFromKafka(Long clusterPhyId, String groupName) throws NotExistException, AdminOperateException {
AdminClient adminClient = kafkaAdminClient.getClient(clusterPhyId);
try {
@@ -117,40 +161,12 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
}
@Override
- public int replaceDBData(GroupMemberPO groupMemberPO) {
- return groupMemberDAO.replace(groupMemberPO);
- }
-
- @Override
- public void batchReplace(List newGroupMemberList) {
- if (newGroupMemberList == null || newGroupMemberList.isEmpty()) {
- return;
- }
-
- Long clusterPhyId = newGroupMemberList.get(0).getClusterPhyId();
- if (clusterPhyId == null) {
- return;
- }
-
- List dbGroupMemberList = listGroup(clusterPhyId);
-
-
- Map dbGroupMemberMap = dbGroupMemberList.stream().collect(Collectors.toMap(elem -> elem.getGroupName() + elem.getTopicName(), Function.identity()));
- for (GroupMemberPO groupMemberPO : newGroupMemberList) {
- GroupMemberPO po = dbGroupMemberMap.remove(groupMemberPO.getGroupName() + groupMemberPO.getTopicName());
- try {
- if (po != null) {
- groupMemberPO.setId(po.getId());
- groupMemberDAO.updateById(groupMemberPO);
- } else {
- groupMemberDAO.insert(groupMemberPO);
- }
- } catch (Exception e) {
- log.error("method=batchReplace||clusterPhyId={}||groupName={}||errMsg=exception", clusterPhyId, groupMemberPO.getGroupName(), e);
- }
-
- }
+ public void batchReplaceGroupsAndMembers(Long clusterPhyId, List newGroupList, long updateTime) {
+ // 更新Group信息
+ this.batchReplaceGroups(clusterPhyId, newGroupList, updateTime);
+ // 更新Group-Topic信息
+ this.batchReplaceGroupMembers(clusterPhyId, newGroupList, updateTime);
}
@Override
@@ -176,14 +192,6 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
return groupMemberDAO.selectList(lambdaQueryWrapper);
}
- @Override
- public List listGroup(Long clusterPhyId) {
- LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
- lambdaQueryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
-
- return groupMemberDAO.selectList(lambdaQueryWrapper);
- }
-
@Override
public PaginationResult pagingGroupMembers(Long clusterPhyId,
String topicName,
@@ -208,8 +216,33 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
return PaginationResult.buildSuc(iPage.getRecords(), iPage);
}
+ @Override
+ public Group getGroupFromDB(Long clusterPhyId, String groupName) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+ lambdaQueryWrapper.eq(GroupPO::getName, groupName);
+
+ GroupPO groupPO = groupDAO.selectOne(lambdaQueryWrapper);
+ return GroupConverter.convert2Group(groupPO);
+ }
+
+ @Override
+ public List listClusterGroups(Long clusterPhyId) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+
+ return groupDAO.selectList(lambdaQueryWrapper).stream().map(elem -> GroupConverter.convert2Group(elem)).collect(Collectors.toList());
+ }
+
@Override
public int deleteByUpdateTimeBeforeInDB(Long clusterPhyId, Date beforeTime) {
+ // 删除过期Group信息
+ LambdaQueryWrapper groupPOLambdaQueryWrapper = new LambdaQueryWrapper<>();
+ groupPOLambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+ groupPOLambdaQueryWrapper.le(GroupPO::getUpdateTime, beforeTime);
+ groupDAO.delete(groupPOLambdaQueryWrapper);
+
+ // 删除过期GroupMember信息
LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
queryWrapper.le(GroupMemberPO::getUpdateTime, beforeTime);
@@ -218,17 +251,19 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
@Override
public List getGroupsFromDB(Long clusterPhyId) {
- LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
- queryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
- List poList = groupMemberDAO.selectList(queryWrapper);
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+
+ List poList = groupDAO.selectList(lambdaQueryWrapper);
if (poList == null) {
poList = new ArrayList<>();
}
- return new ArrayList<>(poList.stream().map(elem -> elem.getGroupName()).collect(Collectors.toSet()));
+
+ return new ArrayList<>(poList.stream().map(elem -> elem.getName()).collect(Collectors.toSet()));
}
@Override
- public GroupMemberPO getGroupFromDB(Long clusterPhyId, String groupName, String topicName) {
+ public GroupMemberPO getGroupTopicFromDB(Long clusterPhyId, String groupName, String topicName) {
LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
queryWrapper.eq(GroupMemberPO::getTopicName, topicName);
@@ -239,28 +274,19 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
@Override
public Integer calGroupCount(Long clusterPhyId) {
- LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
- queryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
- List poList = groupMemberDAO.selectList(queryWrapper);
- if (poList == null) {
- poList = new ArrayList<>();
- }
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
- return poList.stream().map(elem -> elem.getGroupName()).collect(Collectors.toSet()).size();
+ return groupDAO.selectCount(lambdaQueryWrapper);
}
@Override
public Integer calGroupStatCount(Long clusterPhyId, GroupStateEnum stateEnum) {
- LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
- queryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
- queryWrapper.eq(GroupMemberPO::getState, stateEnum.getState());
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+ lambdaQueryWrapper.eq(GroupPO::getState, stateEnum.getState());
- List poList = groupMemberDAO.selectList(queryWrapper);
- if (poList == null) {
- poList = new ArrayList<>();
- }
-
- return poList.stream().map(elem -> elem.getGroupName()).collect(Collectors.toSet()).size();
+ return groupDAO.selectCount(lambdaQueryWrapper);
}
@Override
@@ -303,4 +329,74 @@ public class GroupServiceImpl extends BaseVersionControlService implements Group
/**************************************************** private method ****************************************************/
+ private void batchReplaceGroupMembers(Long clusterPhyId, List newGroupList, long updateTime) {
+ if (ValidateUtils.isEmptyList(newGroupList)) {
+ return;
+ }
+
+ List dbPOList = this.listClusterGroupsMemberPO(clusterPhyId);
+ Map dbPOMap = dbPOList.stream().collect(Collectors.toMap(elem -> elem.getGroupName() + elem.getTopicName(), Function.identity()));
+
+ for (Group group: newGroupList) {
+ for (GroupTopicMember member : group.getTopicMembers()) {
+ try {
+ GroupMemberPO newPO = new GroupMemberPO(clusterPhyId, member.getTopicName(), group.getName(), group.getState().getState(), member.getMemberCount(), new Date(updateTime));
+
+ GroupMemberPO dbPO = dbPOMap.remove(newPO.getGroupName() + newPO.getTopicName());
+ if (dbPO != null) {
+ newPO.setId(dbPO.getId());
+ groupMemberDAO.updateById(newPO);
+ continue;
+ }
+
+ groupMemberDAO.insert(newPO);
+ } catch (Exception e) {
+ log.error(
+ "method=batchReplaceGroupMembers||clusterPhyId={}||groupName={}||topicName={}||errMsg=exception",
+ clusterPhyId, group.getName(), member.getTopicName(), e
+ );
+ }
+ }
+ }
+ }
+
+ private void batchReplaceGroups(Long clusterPhyId, List newGroupList, long updateTime) {
+ if (ValidateUtils.isEmptyList(newGroupList)) {
+ return;
+ }
+
+ List dbGroupList = this.listClusterGroupsPO(clusterPhyId);
+ Map dbGroupMap = dbGroupList.stream().collect(Collectors.toMap(elem -> elem.getName(), Function.identity()));
+
+ for (Group newGroup: newGroupList) {
+ try {
+ GroupPO newPO = GroupConverter.convert2GroupPO(newGroup);
+ newPO.setUpdateTime(new Date(updateTime));
+
+ GroupPO dbPO = dbGroupMap.remove(newGroup.getName());
+ if (dbPO != null) {
+ newPO.setId(dbPO.getId());
+ groupDAO.updateById(newPO);
+ continue;
+ }
+
+ groupDAO.insert(newPO);
+ } catch (Exception e) {
+ log.error("method=batchGroupReplace||clusterPhyId={}||groupName={}||errMsg=exception", clusterPhyId, newGroup.getName(), e);
+ }
+ }
+ }
+
+ private List listClusterGroupsPO(Long clusterPhyId) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupPO::getClusterPhyId, clusterPhyId);
+ return groupDAO.selectList(lambdaQueryWrapper);
+ }
+
+ private List listClusterGroupsMemberPO(Long clusterPhyId) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(GroupMemberPO::getClusterPhyId, clusterPhyId);
+
+ return groupMemberDAO.selectList(lambdaQueryWrapper);
+ }
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/HealthCheckResultService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/HealthCheckResultService.java
index 19cb292a..6aaddcdb 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/HealthCheckResultService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/HealthCheckResultService.java
@@ -20,4 +20,6 @@ public interface HealthCheckResultService {
List getResHealthCheckResult(Long clusterPhyId, Integer dimension, String resNme);
Map getClusterHealthConfig(Long clusterPhyId);
+
+ void batchReplace(Long clusterPhyId, List healthCheckResults);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/impl/HealthCheckResultServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/impl/HealthCheckResultServiceImpl.java
index 09c30a1b..cad2f396 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/impl/HealthCheckResultServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checkresult/impl/HealthCheckResultServiceImpl.java
@@ -3,10 +3,12 @@ package com.xiaojukeji.know.streaming.km.core.service.health.checkresult.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.didiglobal.logi.log.ILog;
import com.didiglobal.logi.log.LogFactory;
+import com.google.common.collect.Lists;
import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
import com.xiaojukeji.know.streaming.km.common.bean.po.config.PlatformClusterConfigPO;
import com.xiaojukeji.know.streaming.km.common.bean.po.health.HealthCheckResultPO;
+import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import com.xiaojukeji.know.streaming.km.common.enums.config.ConfigGroupEnum;
import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckNameEnum;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -88,4 +90,17 @@ public class HealthCheckResultServiceImpl implements HealthCheckResultService {
}
return configMap;
}
+
+ @Override
+ public void batchReplace(Long clusterPhyId, List healthCheckResults) {
+ List> healthCheckResultPartitions = Lists.partition(healthCheckResults, Constant.PER_BATCH_MAX_VALUE);
+ for (List checkResultPartition : healthCheckResultPartitions) {
+ List healthCheckResultPos = ConvertUtil.list2List(checkResultPartition, HealthCheckResultPO.class);
+ try {
+ healthCheckResultDAO.batchReplace(healthCheckResultPos);
+ } catch (Exception e) {
+ log.error("method=batchReplace||clusterPhyId={}||checkResultList={}||errMsg=exception!", clusterPhyId, healthCheckResultPos, e);
+ }
+ }
+ }
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/HealthScoreService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/HealthScoreService.java
index 5997edec..48f8933b 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/HealthScoreService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/HealthScoreService.java
@@ -15,35 +15,16 @@ public interface HealthScoreService {
* @param clusterPhyId 集群ID
* @return
*/
+ @Deprecated
ClusterMetrics calClusterHealthScore(Long clusterPhyId);
- /**
- * 获取集群Topics健康分指标
- * @param clusterPhyId 集群ID
- * @return
- */
- ClusterMetrics calClusterTopicsHealthScore(Long clusterPhyId);
-
- /**
- * 获取集群Brokers健康分指标
- * @param clusterPhyId 集群ID
- * @return
- */
- ClusterMetrics calClusterBrokersHealthScore(Long clusterPhyId);
-
- /**
- * 获取集群Groups健康分指标
- * @param clusterPhyId 集群ID
- * @return
- */
- ClusterMetrics calClusterGroupsHealthScore(Long clusterPhyId);
-
/**
* 获取集群健康分指标
* @param clusterPhyId 集群ID
* @param topicName Topic名称
* @return
*/
+ @Deprecated
TopicMetrics calTopicHealthScore(Long clusterPhyId, String topicName);
/**
@@ -52,6 +33,7 @@ public interface HealthScoreService {
* @param brokerId brokerId
* @return
*/
+ @Deprecated
BrokerMetrics calBrokerHealthScore(Long clusterPhyId, Integer brokerId);
/**
@@ -60,6 +42,7 @@ public interface HealthScoreService {
* @param groupName group名称
* @return
*/
+ @Deprecated
GroupMetrics calGroupHealthScore(Long clusterPhyId, String groupName);
/**
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/impl/HealthScoreServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/impl/HealthScoreServiceImpl.java
index c443c3bb..6ba01bb9 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/impl/HealthScoreServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/score/impl/HealthScoreServiceImpl.java
@@ -136,60 +136,6 @@ public class HealthScoreServiceImpl implements HealthScoreService {
return metrics;
}
- @Override
- public ClusterMetrics calClusterTopicsHealthScore(Long clusterPhyId) {
- List healthScoreResultList = this.getDimensionHealthScoreResult(clusterPhyId, HealthCheckDimensionEnum.TOPIC);
-
- ClusterMetrics metrics = new ClusterMetrics(clusterPhyId);
- if (ValidateUtils.isEmptyList(healthScoreResultList)) {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_TOPICS, Constant.MIN_HEALTH_SCORE);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_TOPICS, 0.0f);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_TOPICS, 0.0f);
- } else {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_TOPICS, Math.max(this.getDimensionHealthScore(healthScoreResultList), Constant.MIN_HEALTH_SCORE));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_TOPICS, getHealthCheckPassed(healthScoreResultList));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_TOPICS, Float.valueOf(healthScoreResultList.size()));
- }
-
- return metrics;
- }
-
- @Override
- public ClusterMetrics calClusterBrokersHealthScore(Long clusterPhyId) {
- List healthScoreResultList = this.getDimensionHealthScoreResult(clusterPhyId, HealthCheckDimensionEnum.BROKER);
-
- ClusterMetrics metrics = new ClusterMetrics(clusterPhyId);
- if (ValidateUtils.isEmptyList(healthScoreResultList)) {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_BROKERS, Constant.MIN_HEALTH_SCORE);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_BROKERS, 0.0f);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_BROKERS, 0.0f);
- } else {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_BROKERS, Math.max(this.getDimensionHealthScore(healthScoreResultList), Constant.MIN_HEALTH_SCORE));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_BROKERS, getHealthCheckPassed(healthScoreResultList));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_BROKERS, Float.valueOf(healthScoreResultList.size()));
- }
-
- return metrics;
- }
-
- @Override
- public ClusterMetrics calClusterGroupsHealthScore(Long clusterPhyId) {
- List healthScoreResultList = this.getDimensionHealthScoreResult(clusterPhyId, HealthCheckDimensionEnum.GROUP);
-
- ClusterMetrics metrics = new ClusterMetrics(clusterPhyId);
- if (ValidateUtils.isEmptyList(healthScoreResultList)) {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_GROUPS, Constant.MIN_HEALTH_SCORE);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_GROUPS, 0.0f);
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_GROUPS, 0.0f);
- } else {
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_SCORE_GROUPS, Math.max(this.getDimensionHealthScore(healthScoreResultList), Constant.MIN_HEALTH_SCORE));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_PASSED_GROUPS, this.getHealthCheckPassed(healthScoreResultList));
- metrics.getMetrics().put(CLUSTER_METRIC_HEALTH_CHECK_TOTAL_GROUPS, Float.valueOf(healthScoreResultList.size()));
- }
-
- return metrics;
- }
-
@Override
public TopicMetrics calTopicHealthScore(Long clusterPhyId, String topicName) {
List healthScoreResultList = this.getResHealthScoreResult(clusterPhyId, HealthCheckDimensionEnum.TOPIC.getDimension(), topicName);
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/ReplicaMetricService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/ReplicaMetricService.java
index 987303f8..c0a44586 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/ReplicaMetricService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/ReplicaMetricService.java
@@ -13,12 +13,14 @@ public interface ReplicaMetricService {
* 从kafka中采集指标
*/
Result collectReplicaMetricsFromKafka(Long clusterId, String topic, Integer partitionId, Integer brokerId, String metric);
- Result collectReplicaMetricsFromKafkaWithCache(Long clusterPhyId, String topic, Integer brokerId, Integer partitionId, String metric);
+ Result collectReplicaMetricsFromKafka(Long clusterId, String topicName, Integer partitionId, Integer brokerId, List metricNameList);
/**
* 从ES中获取指标
*/
+ @Deprecated
Result> getMetricPointsFromES(Long clusterPhyId, Integer brokerId, String topicName, Integer partitionId, MetricDTO dto);
+ @Deprecated
Result getLatestMetricsFromES(Long clusterPhyId, Integer brokerId, String topicName, Integer partitionId, List metricNames);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/impl/ReplicaMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/impl/ReplicaMetricServiceImpl.java
index 460e6520..848c8601 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/impl/ReplicaMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/replica/impl/ReplicaMetricServiceImpl.java
@@ -17,7 +17,6 @@ import com.xiaojukeji.know.streaming.km.common.jmx.JmxConnectorWrap;
import com.xiaojukeji.know.streaming.km.common.utils.BeanUtil;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
-import com.xiaojukeji.know.streaming.km.core.cache.CollectedMetricsLocalCache;
import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService;
import com.xiaojukeji.know.streaming.km.core.service.replica.ReplicaMetricService;
import com.xiaojukeji.know.streaming.km.core.service.version.BaseMetricService;
@@ -77,32 +76,36 @@ public class ReplicaMetricServiceImpl extends BaseMetricService implements Repli
}
@Override
- public Result collectReplicaMetricsFromKafkaWithCache(Long clusterPhyId,
- String topic,
- Integer brokerId,
- Integer partitionId,
- String metric) {
- String replicaMetricsKey = CollectedMetricsLocalCache.genReplicaMetricCacheKey(clusterPhyId, brokerId, topic, partitionId, metric);
+ public Result collectReplicaMetricsFromKafka(Long clusterId, String topicName, Integer partitionId, Integer brokerId, List metricNameList) {
+ ReplicationMetrics metrics = new ReplicationMetrics(clusterId, topicName, brokerId, partitionId);
+ for (String metricName: metricNameList) {
+ try {
+ if (metrics.getMetrics().containsKey(metricName)) {
+ continue;
+ }
- Float keyValue = CollectedMetricsLocalCache.getReplicaMetrics(replicaMetricsKey);
- if(null != keyValue){
- ReplicationMetrics replicationMetrics = new ReplicationMetrics(clusterPhyId, topic, partitionId, brokerId);
- replicationMetrics.putMetric(metric, keyValue);
- return Result.buildSuc(replicationMetrics);
+ Result ret = this.collectReplicaMetricsFromKafka(
+ clusterId,
+ metrics.getTopic(),
+ metrics.getBrokerId(),
+ metrics.getPartitionId(),
+ metricName
+ );
+
+ if (null == ret || ret.failed() || null == ret.getData()) {
+ continue;
+ }
+
+ metrics.putMetric(ret.getData().getMetrics());
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=collectReplicaMetricsFromKafka||clusterPhyId={}||topicName={}||partition={}||brokerId={}||metricName={}||errMsg=exception!",
+ clusterId, topicName, partitionId, brokerId, e
+ );
+ }
}
- Result ret = collectReplicaMetricsFromKafka(clusterPhyId, topic, partitionId, brokerId, metric);
- if(null == ret || ret.failed() || null == ret.getData()){return ret;}
-
- // 更新cache
- ret.getData().getMetrics().entrySet().stream().forEach(
- metricNameAndValueEntry -> CollectedMetricsLocalCache.putReplicaMetrics(
- replicaMetricsKey,
- metricNameAndValueEntry.getValue()
- )
- );
-
- return ret;
+ return Result.buildSuc(metrics);
}
@Override
@@ -167,8 +170,8 @@ public class ReplicaMetricServiceImpl extends BaseMetricService implements Repli
Integer brokerId = metricParam.getBrokerId();
Integer partitionId = metricParam.getPartitionId();
- Result endRet = this.collectReplicaMetricsFromKafkaWithCache(clusterId, topic, brokerId, partitionId, REPLICATION_METRIC_LOG_END_OFFSET);
- Result startRet = this.collectReplicaMetricsFromKafkaWithCache(clusterId, topic, brokerId, partitionId, REPLICATION_METRIC_LOG_START_OFFSET);
+ Result endRet = this.collectReplicaMetricsFromKafka(clusterId, topic, brokerId, partitionId, REPLICATION_METRIC_LOG_END_OFFSET);
+ Result startRet = this.collectReplicaMetricsFromKafka(clusterId, topic, brokerId, partitionId, REPLICATION_METRIC_LOG_START_OFFSET);
ReplicationMetrics replicationMetrics = new ReplicationMetrics(clusterId, topic, brokerId, partitionId);
if(null != endRet && endRet.successful() && null != startRet && startRet.successful()){
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ClusterMetricVersionItems.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ClusterMetricVersionItems.java
index 53b98479..00a5e0cd 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ClusterMetricVersionItems.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ClusterMetricVersionItems.java
@@ -20,22 +20,35 @@ import static com.xiaojukeji.know.streaming.km.core.service.cluster.impl.Cluster
@Component
public class ClusterMetricVersionItems extends BaseMetricVersionMetric {
+ /**
+ * 健康分
+ */
public static final String CLUSTER_METRIC_HEALTH_SCORE = "HealthScore";
+ public static final String CLUSTER_METRIC_HEALTH_SCORE_TOPICS = "HealthScore_Topics";
+ public static final String CLUSTER_METRIC_HEALTH_SCORE_BROKERS = "HealthScore_Brokers";
+ public static final String CLUSTER_METRIC_HEALTH_SCORE_GROUPS = "HealthScore_Groups";
+ public static final String CLUSTER_METRIC_HEALTH_SCORE_CLUSTER = "HealthScore_Cluster";
+
+ /**
+ * 健康巡检
+ */
public static final String CLUSTER_METRIC_HEALTH_CHECK_PASSED = "HealthCheckPassed";
public static final String CLUSTER_METRIC_HEALTH_CHECK_TOTAL = "HealthCheckTotal";
- public static final String CLUSTER_METRIC_HEALTH_SCORE_TOPICS = "HealthScore_Topics";
+
public static final String CLUSTER_METRIC_HEALTH_CHECK_PASSED_TOPICS = "HealthCheckPassed_Topics";
public static final String CLUSTER_METRIC_HEALTH_CHECK_TOTAL_TOPICS = "HealthCheckTotal_Topics";
- public static final String CLUSTER_METRIC_HEALTH_SCORE_BROKERS = "HealthScore_Brokers";
+
public static final String CLUSTER_METRIC_HEALTH_CHECK_PASSED_BROKERS = "HealthCheckPassed_Brokers";
public static final String CLUSTER_METRIC_HEALTH_CHECK_TOTAL_BROKERS = "HealthCheckTotal_Brokers";
- public static final String CLUSTER_METRIC_HEALTH_SCORE_GROUPS = "HealthScore_Groups";
+
public static final String CLUSTER_METRIC_HEALTH_CHECK_PASSED_GROUPS = "HealthCheckPassed_Groups";
public static final String CLUSTER_METRIC_HEALTH_CHECK_TOTAL_GROUPS = "HealthCheckTotal_Groups";
- public static final String CLUSTER_METRIC_HEALTH_SCORE_CLUSTER = "HealthScore_Cluster";
+
public static final String CLUSTER_METRIC_HEALTH_CHECK_PASSED_CLUSTER = "HealthCheckPassed_Cluster";
public static final String CLUSTER_METRIC_HEALTH_CHECK_TOTAL_CLUSTER = "HealthCheckTotal_Cluster";
+
+
public static final String CLUSTER_METRIC_TOTAL_REQ_QUEUE_SIZE = "TotalRequestQueueSize";
public static final String CLUSTER_METRIC_TOTAL_RES_QUEUE_SIZE = "TotalResponseQueueSize";
public static final String CLUSTER_METRIC_EVENT_QUEUE_SIZE = "EventQueueSize";
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ZookeeperMetricVersionItems.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ZookeeperMetricVersionItems.java
new file mode 100644
index 00000000..9b0d4d2b
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/ZookeeperMetricVersionItems.java
@@ -0,0 +1,141 @@
+package com.xiaojukeji.know.streaming.km.core.service.version.metrics;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionMetricControlItem;
+import com.xiaojukeji.know.streaming.km.common.constant.Constant;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionMetricControlItem.*;
+import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum.METRIC_ZOOKEEPER;
+import static com.xiaojukeji.know.streaming.km.common.jmx.JmxAttribute.*;
+import static com.xiaojukeji.know.streaming.km.common.jmx.JmxName.*;
+import static com.xiaojukeji.know.streaming.km.core.service.zookeeper.impl.ZookeeperMetricServiceImpl.*;
+
+@Component
+public class ZookeeperMetricVersionItems extends BaseMetricVersionMetric {
+
+ /**
+ * 性能
+ */
+ public static final String ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY = "AvgRequestLatency";
+ public static final String ZOOKEEPER_METRIC_MIN_REQUEST_LATENCY = "MinRequestLatency";
+ public static final String ZOOKEEPER_METRIC_MAX_REQUEST_LATENCY = "MaxRequestLatency";
+ public static final String ZOOKEEPER_METRIC_OUTSTANDING_REQUESTS = "OutstandingRequests";
+ public static final String ZOOKEEPER_METRIC_NODE_COUNT = "NodeCount";
+ public static final String ZOOKEEPER_METRIC_WATCH_COUNT = "WatchCount";
+ public static final String ZOOKEEPER_METRIC_NUM_ALIVE_CONNECTIONS = "NumAliveConnections";
+ public static final String ZOOKEEPER_METRIC_PACKETS_RECEIVED = "PacketsReceived";
+ public static final String ZOOKEEPER_METRIC_PACKETS_SENT = "PacketsSent";
+ public static final String ZOOKEEPER_METRIC_EPHEMERALS_COUNT = "EphemeralsCount";
+ public static final String ZOOKEEPER_METRIC_APPROXIMATE_DATA_SIZE = "ApproximateDataSize";
+ public static final String ZOOKEEPER_METRIC_OPEN_FILE_DESCRIPTOR_COUNT = "OpenFileDescriptorCount";
+ public static final String ZOOKEEPER_METRIC_MAX_FILE_DESCRIPTOR_COUNT = "MaxFileDescriptorCount";
+
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_DISCONNECTS_PER_SEC = "KafkaZKDisconnectsPerSec";
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_SYNC_CONNECTS_PER_SEC = "KafkaZKSyncConnectsPerSec";
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_99TH = "KafkaZKRequestLatencyMs_99thPercentile";
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MAX = "KafkaZKRequestLatencyMs_Max";
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MEAN = "KafkaZKRequestLatencyMs_Mean";
+ public static final String ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MIN = "KafkaZKRequestLatencyMs_Min";
+
+
+ public static final String ZOOKEEPER_METRIC_COLLECT_COST_TIME = Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME;
+
+ @Override
+ public int versionItemType() {
+ return METRIC_ZOOKEEPER.getCode();
+ }
+
+ @Override
+ public List init(){
+ List items = new ArrayList<>();
+
+ // 性能指标
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY).unit("ms").desc("平均响应延迟").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_MIN_REQUEST_LATENCY).unit("ms").desc("最小响应延迟").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_MAX_REQUEST_LATENCY).unit("ms").desc("最大响应延迟").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_OUTSTANDING_REQUESTS).unit("个").desc("堆积请求数").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_NODE_COUNT).unit("个").desc("ZNode数量").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_WATCH_COUNT).unit("个").desc("Watch数量").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_NUM_ALIVE_CONNECTIONS).unit("个").desc("客户端连接数量").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_PACKETS_RECEIVED).unit("个").desc("接受包的数量").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_PACKETS_SENT).unit("个").desc("发送包的数量").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_EPHEMERALS_COUNT).unit("个").desc("临时节点数").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_APPROXIMATE_DATA_SIZE).unit("byte").desc("文件大小(近似值)").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_OPEN_FILE_DESCRIPTOR_COUNT).unit("个").desc("已打开的文件描述符数").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_MAX_FILE_DESCRIPTOR_COUNT).unit("个").desc("允许打开的最大文件描述符数").category(CATEGORY_PERFORMANCE)
+ .extendMethod(ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD));
+
+ // JMX指标
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_99TH).unit("ms").desc("ZK请求99分位延迟").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_REQUEST_LATENCY_MS ).jmxAttribute(PERCENTILE_99)));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MAX).unit("ms").desc("ZK请求最大延迟").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_REQUEST_LATENCY_MS ).jmxAttribute(MAX)));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MIN).unit("ms").desc("ZK请求最小延迟").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_REQUEST_LATENCY_MS ).jmxAttribute(MIN)));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_REQUEST_LATENCY_MEAN).unit("ms").desc("ZK请求平均延迟").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_REQUEST_LATENCY_MS ).jmxAttribute(MEAN)));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_DISCONNECTS_PER_SEC).unit("个").desc("断开连接数").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_DISCONNECTORS_PER_SEC ).jmxAttribute(RATE_MIN_1)));
+
+ items.add(buildAllVersionsItem()
+ .name(ZOOKEEPER_METRIC_KAFKA_ZK_SYNC_CONNECTS_PER_SEC).unit("个").desc("同步连接数").category(CATEGORY_CLIENT)
+ .extend( buildJMXMethodExtend( ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX )
+ .jmxObjectName( JMX_ZK_SYNC_CONNECTS_PER_SEC ).jmxAttribute(RATE_MIN_1)));
+ return items;
+ }
+}
+
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZnodeService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZnodeService.java
new file mode 100644
index 00000000..43e7744d
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZnodeService.java
@@ -0,0 +1,13 @@
+package com.xiaojukeji.know.streaming.km.core.service.zookeeper;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.Znode;
+
+import java.util.List;
+
+public interface ZnodeService {
+
+ Result> listZnodeChildren(Long clusterPhyId, String path, String keyword);
+
+ Result 
## Star History
diff --git a/Releases_Notes.md b/Releases_Notes.md
index f453b582..a3df746f 100644
--- a/Releases_Notes.md
+++ b/Releases_Notes.md
@@ -1,4 +1,36 @@
+## v3.0.1
+
+**Bug修复**
+- 修复重置 Group Offset 时，提示信息中缺少 Dead 状态也可进行重置的信息；
+- 修复 Ldap 某个属性不存在时，会直接抛出空指针导致登陆失败的问题；
+- 修复集群 Topic 列表页，健康分详情信息中，检查时间展示错误的问题；
+- 修复更新健康检查结果时，出现死锁的问题；
+- 修复 Replica 索引模版错误的问题；
+- 修复 FAQ 文档中的错误链接；
+- 修复 Broker 的 TopN 指标不存在时，页面数据不展示的问题；
+- 修复 Group 详情页，图表时间范围选择不生效的问题；
+
+
+**体验优化**
+- 集群 Group 列表按照 Group 维度进行展示；
+- 优化避免因 ES 中该指标不存在，导致日志中出现大量空指针的问题；
+- 优化全局 Message & Notification 展示效果；
+- 优化 Topic 扩分区名称 & 描述展示；
+
+
+**新增**
+- Broker 列表页面，新增 JMX 是否成功连接的信息；
+
+
+**ZK 部分(未完全发布)**
+- 后端补充 Kafka ZK 指标采集，Kafka ZK 信息获取相关功能；
+- 增加本地缓存，避免同一采集周期内 ZK 指标重复采集；
+- 增加 ZK 节点采集失败跳过策略，避免不断对存在问题的节点不断尝试；
+- 修复 zkAvgLatency 指标转 Long 时抛出异常问题；
+- 修复 ks_km_zookeeper 表中，role 字段类型错误问题；
+
+---
## v3.0.0
@@ -25,7 +57,7 @@
- 集群信息中，新增 Kafka 集群运行模式字段
- 新增 docker-compose 的部署方式
-
+---
## v3.0.0-beta.3
diff --git a/bin/init_es_template.sh b/bin/init_es_template.sh
index e6beba96..86fcfb66 100644
--- a/bin/init_es_template.sh
+++ b/bin/init_es_template.sh
@@ -439,7 +439,7 @@ curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: appl
curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: application/json' http://${esaddr}:${port}/_template/ks_kafka_replication_metric -d '{
"order" : 10,
"index_patterns" : [
- "ks_kafka_partition_metric*"
+ "ks_kafka_replication_metric*"
],
"settings" : {
"index" : {
@@ -500,30 +500,7 @@ curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: appl
}
},
"aliases" : { }
- }[root@10-255-0-23 template]# cat ks_kafka_replication_metric
-PUT _template/ks_kafka_replication_metric
-{
- "order" : 10,
- "index_patterns" : [
- "ks_kafka_replication_metric*"
- ],
- "settings" : {
- "index" : {
- "number_of_shards" : "10"
- }
- },
- "mappings" : {
- "properties" : {
- "timestamp" : {
- "format" : "yyyy-MM-dd HH:mm:ss Z||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS Z||yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss,SSS||yyyy/MM/dd HH:mm:ss||yyyy-MM-dd HH:mm:ss,SSS Z||yyyy/MM/dd HH:mm:ss,SSS Z||epoch_millis",
- "index" : true,
- "type" : "date",
- "doc_values" : true
- }
- }
- },
- "aliases" : { }
- }'
+ }'
curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: application/json' http://${esaddr}:${port}/_template/ks_kafka_topic_metric -d '{
"order" : 10,
@@ -640,7 +617,92 @@ curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: appl
}
},
"aliases" : { }
- }'
+ }'
+
+curl -s -o /dev/null -X POST -H 'cache-control: no-cache' -H 'content-type: application/json' http://${SERVER_ES_ADDRESS}/_template/ks_kafka_zookeeper_metric -d '{
+ "order" : 10,
+ "index_patterns" : [
+ "ks_kafka_zookeeper_metric*"
+ ],
+ "settings" : {
+ "index" : {
+ "number_of_shards" : "10"
+ }
+ },
+ "mappings" : {
+ "properties" : {
+ "routingValue" : {
+ "type" : "text",
+ "fields" : {
+ "keyword" : {
+ "ignore_above" : 256,
+ "type" : "keyword"
+ }
+ }
+ },
+ "clusterPhyId" : {
+ "type" : "long"
+ },
+ "metrics" : {
+ "properties" : {
+ "AvgRequestLatency" : {
+ "type" : "double"
+ },
+ "MinRequestLatency" : {
+ "type" : "double"
+ },
+ "MaxRequestLatency" : {
+ "type" : "double"
+ },
+ "OutstandingRequests" : {
+ "type" : "double"
+ },
+ "NodeCount" : {
+ "type" : "double"
+ },
+ "WatchCount" : {
+ "type" : "double"
+ },
+ "NumAliveConnections" : {
+ "type" : "double"
+ },
+ "PacketsReceived" : {
+ "type" : "double"
+ },
+ "PacketsSent" : {
+ "type" : "double"
+ },
+ "EphemeralsCount" : {
+ "type" : "double"
+ },
+ "ApproximateDataSize" : {
+ "type" : "double"
+ },
+ "OpenFileDescriptorCount" : {
+ "type" : "double"
+ },
+ "MaxFileDescriptorCount" : {
+ "type" : "double"
+ }
+ }
+ },
+ "key" : {
+ "type" : "text",
+ "fields" : {
+ "keyword" : {
+ "ignore_above" : 256,
+ "type" : "keyword"
+ }
+ }
+ },
+ "timestamp" : {
+ "format" : "yyyy-MM-dd HH:mm:ss Z||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS Z||yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss,SSS||yyyy/MM/dd HH:mm:ss||yyyy-MM-dd HH:mm:ss,SSS Z||yyyy/MM/dd HH:mm:ss,SSS Z||epoch_millis",
+ "type" : "date"
+ }
+ }
+ },
+ "aliases" : { }
+ }'
for i in {0..6};
do
@@ -650,6 +712,7 @@ do
curl -s -o /dev/null -X PUT http://${esaddr}:${port}/ks_kafka_group_metric${logdate} && \
curl -s -o /dev/null -X PUT http://${esaddr}:${port}/ks_kafka_partition_metric${logdate} && \
curl -s -o /dev/null -X PUT http://${esaddr}:${port}/ks_kafka_replication_metric${logdate} && \
+ curl -s -o /dev/null -X PUT http://${esaddr}:${port}/ks_kafka_zookeeper_metric${logdate} && \
curl -s -o /dev/null -X PUT http://${esaddr}:${port}/ks_kafka_topic_metric${logdate} || \
exit 2
-done
\ No newline at end of file
+done
diff --git a/docs/contributer_guide/代码规范.md b/docs/contributer_guide/代码规范.md
new file mode 100644
index 00000000..be0bcfd2
--- /dev/null
+++ b/docs/contributer_guide/代码规范.md
@@ -0,0 +1 @@
+TODO.
\ No newline at end of file
diff --git a/docs/contributer_guide/开发者名单.md b/docs/contributer_guide/开发者名单.md
new file mode 100644
index 00000000..3f2c708d
--- /dev/null
+++ b/docs/contributer_guide/开发者名单.md
@@ -0,0 +1,6 @@
+
+开源贡献者证书发放名单(定期更新)
+
+
+贡献者名单请看：[贡献者名单](https://doc.knowstreaming.com/product/10-contribution#106-贡献者名单)
+
diff --git a/docs/contributer_guide/贡献流程.md b/docs/contributer_guide/贡献流程.md
new file mode 100644
index 00000000..42679379
--- /dev/null
+++ b/docs/contributer_guide/贡献流程.md
@@ -0,0 +1,6 @@
+
+
+