ShowProcessListCommand.java

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.doris.nereids.trees.plans.commands;

import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.SchemaTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.common.ClientPool;
import org.apache.doris.common.Pair;
import org.apache.doris.common.proc.FrontendsProcNode;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ShowResultSet;
import org.apache.doris.qe.ShowResultSetMetaData;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.thrift.FrontendService;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TShowProcessListRequest;
import org.apache.doris.thrift.TShowProcessListResult;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.List;
import java.util.Optional;

/**
 * Represents the command for SHOW PROCESSLIST
 */
public class ShowProcessListCommand extends ShowCommand {
    private static final Logger LOG = LogManager.getLogger(ShowProcessListCommand.class);
    private static final ShowResultSetMetaData PROCESSLIST_META_DATA;

    static {
        Table tbl = SchemaTable.TABLE_MAP.get("processlist");
        ShowResultSetMetaData.Builder builder = ShowResultSetMetaData.builder();
        tbl.getBaseSchema().stream().forEach(column -> builder.addColumn(column));
        PROCESSLIST_META_DATA = builder.build();
    }

    private final boolean isFull;

    public ShowProcessListCommand(boolean isFull) {
        super(PlanType.SHOW_PROCESSLIST_COMMAND);
        this.isFull = isFull;
    }

    @Override
    public ShowResultSetMetaData getMetaData() {
        return PROCESSLIST_META_DATA;
    }

    @Override
    public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
        boolean isShowFullSql = isFull;
        boolean isShowAllFe = ConnectContext.get().getSessionVariable().isFetchAllFeForSystemTable();

        List<List<String>> rowSet = Lists.newArrayList();
        List<ConnectContext.ThreadInfo> threadInfos = ctx.getConnectScheduler()
                .listConnection(ctx.getQualifiedUser(), isShowFullSql);
        long nowMs = System.currentTimeMillis();
        for (ConnectContext.ThreadInfo info : threadInfos) {
            rowSet.add(info.toRow(ctx.getConnectionId(), nowMs, Optional.empty()));
        }

        if (isShowAllFe) {
            try {
                TShowProcessListRequest request = new TShowProcessListRequest();
                request.setShowFullSql(isShowFullSql);
                request.setCurrentUserIdent(ConnectContext.get().getCurrentUserIdentity().toThrift());
                List<Pair<String, Integer>> frontends = FrontendsProcNode.getFrontendWithRpcPort(Env.getCurrentEnv(),
                        false);
                FrontendService.Client client = null;
                for (Pair<String, Integer> fe : frontends) {
                    TNetworkAddress thriftAddress = new TNetworkAddress(fe.key(), fe.value());
                    try {
                        client = ClientPool.frontendPool.borrowObject(thriftAddress, 3000);
                    } catch (Exception e) {
                        LOG.warn("Failed to get frontend {} client. exception: {}", fe.key(), e);
                        continue;
                    }

                    boolean isReturnToPool = false;
                    try {
                        TShowProcessListResult result = client.showProcessList(request);
                        if (result.process_list != null && result.process_list.size() > 0) {
                            for (List<String> row : result.process_list) {
                                rowSet.add(fitToColumns(row, PROCESSLIST_META_DATA.getColumnCount()));
                            }
                        }
                        isReturnToPool = true;
                    } catch (Exception e) {
                        LOG.warn("Failed to request processlist to fe: {} . exception: {}", fe.key(), e);
                    } finally {
                        if (isReturnToPool) {
                            ClientPool.frontendPool.returnObject(thriftAddress, client);
                        } else {
                            ClientPool.frontendPool.invalidateObject(thriftAddress, client);
                        }
                    }
                }
            } catch (Throwable t) {
                LOG.warn(" fetch process list from other fe failed, ", t);
            }
        }

        return new ShowResultSet(getMetaData(), rowSet);
    }

    /**
     * A row from another frontend has that frontend's columns, which differ from this one's while the
     * cluster runs two versions: an older frontend's row ends before the columns added since (Protocol),
     * a newer one's has columns this frontend does not know. Fit it to the local column list so the
     * result set stays rectangular: pad with empty strings, drop the surplus.
     */
    @VisibleForTesting
    static List<String> fitToColumns(List<String> row, int columnCount) {
        if (row.size() == columnCount) {
            return row;
        }
        List<String> fitted = Lists.newArrayListWithCapacity(columnCount);
        fitted.addAll(row.subList(0, Math.min(row.size(), columnCount)));
        while (fitted.size() < columnCount) {
            fitted.add("");
        }
        return fitted;
    }

    @Override
    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
        return visitor.visitShowProcessListCommand(this, context);
    }
}