IGNITE-24286 Implement recording sytemViews to perfStart report

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/AttributeToMapVisitor.java

@@ -0,0 +1,82 @@
+/*
+ * 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.ignite.internal.processors.performancestatistics;
+
+import java.util.Map;
+import org.apache.ignite.spi.systemview.view.SystemViewRowAttributeWalker;
+import org.jetbrains.annotations.Nullable;
+
+/** Fullfill {@code data} Map for specific row. */
+class AttributeToMapVisitor implements SystemViewRowAttributeWalker.AttributeWithValueVisitor {
+    /** Map to store data. */
+    private Map<String, String> data;
+
+    /**
+     * Sets map.
+     *
+     * @param data Map to fill.
+     */
+    public void data(Map<String, String> data) {
+        this.data = data;
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> void accept(int idx, String name, Class<T> clazz, @Nullable T val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptBoolean(int idx, String name, boolean val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptChar(int idx, String name, char val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptByte(int idx, String name, byte val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptShort(int idx, String name, short val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptInt(int idx, String name, int val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptLong(int idx, String name, long val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptFloat(int idx, String name, float val) {
+        data.put(name, String.valueOf(val));
+    }
+
+    /** {@inheritDoc} */
+    @Override public void acceptDouble(int idx, String name, double val) {
+        data.put(name, String.valueOf(val));
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java

@@ -55,6 +55,7 @@
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_PROPERTY;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_READS;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_ROWS;
+import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.TASK;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.TX_COMMIT;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.VERSION;
@@ -312,6 +313,34 @@ else if (opType == QUERY) {
 
             return true;
         }
+        else if (opType == SYSTEM_VIEW) {
+            String viewName = readCacheableString(buf);
+            if (viewName == null)
+                return false;
+
+            if (buf.remaining() < 4)
+                return false;
+
+            int attrsNumber = buf.getInt();
+
+            Map<String, String> data = new HashMap<>();
+            for (int i = 0; i < attrsNumber; i++) {
+                String key = readCacheableString(buf);
+                if (key == null)
+                    return false;
+
+                String val = readCacheableString(buf);
+                if (val == null)
+                    return false;
+
+                data.put(key, val);
+            }
+
+            for (PerformanceStatisticsHandler hnd : curHnd)
+                hnd.systemView(nodeId, viewName, data);
+
+            return true;
+        }
         else if (opType == QUERY_READS) {
             if (buf.remaining() < queryReadsRecordSize())
                 return false;

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsWriter.java

@@ -21,8 +21,10 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedByInterruptException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -41,8 +43,10 @@
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.util.GridIntIterator;
 import org.apache.ignite.internal.util.GridIntList;
+import org.apache.ignite.internal.util.lang.IgnitePair;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.internal.util.worker.GridWorker;
+import org.apache.ignite.lang.IgniteBiTuple;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.thread.IgniteThread;
 
@@ -58,6 +62,7 @@
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_PROPERTY;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_READS;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.QUERY_ROWS;
+import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.TASK;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.TX_COMMIT;
 import static org.apache.ignite.internal.processors.performancestatistics.OperationType.TX_ROLLBACK;
@@ -275,6 +280,40 @@ public void query(GridCacheQueryType type, String text, long id, long startTime,
         });
     }
 
+    /**
+     * @param view Name of system view.
+     * @param row  Data from the view.
+     */
+    public void systemView(String view, Map<String, String> row) {
+        boolean cachedName = cacheIfPossible(view);
+
+        int recSize = 1 + (cachedName ? 4 : 4 + view.getBytes().length) + 4;
+
+        List<IgnitePair<IgniteBiTuple<String, Boolean>>> cachedRow = new ArrayList<>();
+        for (Map.Entry<String, String> entry : row.entrySet()) {
+            String key = entry.getKey();
+            String val = entry.getValue();
+
+            IgniteBiTuple<String, Boolean> cachedKey = new IgniteBiTuple<>(key, cacheIfPossible(key));
+            IgniteBiTuple<String, Boolean> cachedVal = new IgniteBiTuple<>(val, cacheIfPossible(val));
+
+            recSize += 1 + (cachedKey.getValue() ? 4 : 4 + key.getBytes().length);
+            recSize += 1 + (cachedVal.getValue() ? 4 : 4 + val.getBytes().length);
+
+            cachedRow.add(new IgnitePair<>(cachedKey, cachedVal));
+        }
+
+        doWrite(SYSTEM_VIEW, recSize, buf -> {
+            writeString(buf, view, cachedName);
+            buf.putInt(row.size());
+
+            for (IgnitePair<IgniteBiTuple<String, Boolean>> pair : cachedRow) {
+                writeString(buf, pair.getKey().getKey(), pair.getKey().getValue());
+                writeString(buf, pair.getValue().getKey(), pair.getValue().getValue());
+            }
+        });
+    }
+
     /**
      * @param type Cache query type.
      * @param queryNodeId Originating node id.

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/OperationType.java

@@ -93,6 +93,9 @@ public enum OperationType {
     /** Custom query property. */
     QUERY_PROPERTY(21),
 
+    /** System view. */
+    SYSTEM_VIEW(22),
+
     /** Version. */
     VERSION(255);
 

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsHandler.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.performancestatistics;
 
+import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
 import org.apache.ignite.internal.util.GridIntList;
@@ -164,4 +165,11 @@ void checkpoint(
      * @param duration Duration in milliseconds.
      */
     void pagesWriteThrottle(UUID nodeId, long endTime, long duration);
+
+    /**
+     * @param id      Node id.
+     * @param name    Name of system view.
+     * @param data    Data from system view.
+     */
+    void systemView(UUID id, String name, Map<String, String> data);
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java

@@ -20,6 +20,9 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.EventListener;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 import java.util.UUID;
 import java.util.function.Consumer;
 import org.apache.ignite.IgniteCheckedException;
@@ -41,6 +44,7 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.systemview.view.SystemView;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.processors.metastorage.DistributedMetaStorage.IGNITE_INTERNAL_KEY_PREFIX;
@@ -81,6 +85,22 @@ public PerformanceStatisticsProcessor(GridKernalContext ctx) {
             if (U.isLocalNodeCoordinator(ctx.discovery()))
                 ctx.cache().cacheDescriptors().values().forEach(desc -> cacheStart(desc.cacheId(), desc.cacheName()));
         });
+
+        registerStateListener(new PerformanceStatisticsStateListener() {
+            private final List<String> ignoredViews = List.of(
+                "baseline.node.attributes",
+                "metrics",
+                "caches",
+                "sql.queries",
+                "nodes");
+
+            @Override public void onStarted() {
+                for (SystemView<?> sysView : ctx.systemView()) {
+                    if (!ignoredViews.contains(sysView.name()))
+                        systemView(sysView);
+                }
+            }
+        });
     }
 
     /** {@inheritDoc} */
@@ -172,6 +192,21 @@ public void query(GridCacheQueryType type, String text, long id, long startTime,
         write(writer -> writer.query(type, text, id, startTime, duration, success));
     }
 
+    /**
+     * @param view System view to extract data from.
+     */
+    public void systemView(SystemView<?> view) {
+        AttributeToMapVisitor visitor = new AttributeToMapVisitor();
+
+        for (Object row : view) {
+            Map<String, String> data = new TreeMap<>();
+            visitor.data(data);
+            ((SystemView<Object>)view).walker().visitAll(row, visitor);
+
+            write(writer -> writer.systemView(view.name(), data));
+        }
+    }
+
     /**
      * @param type Cache query type.
      * @param queryNodeId Originating node id.

modules/core/src/test/java/org/apache/ignite/internal/processors/performancestatistics/AbstractPerformanceStatisticsTest.java

@@ -20,6 +20,7 @@
 import java.io.File;
 import java.lang.management.ThreadInfo;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
@@ -237,6 +238,11 @@ public static class TestHandler implements PerformanceStatisticsHandler {
         @Override public void pagesWriteThrottle(UUID nodeId, long endTime, long duration) {
             // No-op.
         }
+
+        /** {@inheritDoc} */
+        @Override public void systemView(UUID id, String name, Map<String, String> data) {
+            // No-op.
+        }
     }
 
     /** Client type to run load from. */

modules/core/src/test/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsSystemViewTest.java

@@ -0,0 +1,85 @@
+/*
+ * 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.ignite.internal.processors.performancestatistics;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.junit.Test;
+
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/**
+ * Tests performance start with system views.
+ */
+public class PerformanceStatisticsSystemViewTest extends AbstractPerformanceStatisticsTest {
+    /** */
+    private final ListeningTestLogger listeningLog = new ListeningTestLogger();
+
+    /** */
+    private static final List<String> IGNORED_VIEWS = List.of(
+        "baseline.node.attributes",
+        "metrics",
+        "caches",
+        "sql.queries",
+        "nodes");
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        cfg.setCacheConfiguration(defaultCacheConfiguration());
+        cfg.setGridLogger(listeningLog);
+
+        return cfg;
+    }
+
+    /** @throws Exception If failed. */
+    @Test
+    public void testSystemViewCaches() throws Exception {
+        LogListener lsnr = LogListener.matches("Performance statistics writer started.").build();
+        listeningLog.registerListener(lsnr);
+
+        try (IgniteEx igniteEx = startGrid(0)) {
+            startCollectStatistics();
+
+            Set<String> viewsExpected = new HashSet<>();
+            igniteEx.context().systemView().forEach(view -> {
+                if (view.size() > 0 && !IGNORED_VIEWS.contains(view.name()))
+                    viewsExpected.add(view.name());
+            });
+
+            assertTrue("Performance statistics writer did not start.", waitForCondition(lsnr::check, TIMEOUT));
+
+            Set<String> viewsActual = new HashSet<>();
+            stopCollectStatisticsAndRead(new TestHandler() {
+                @Override public void systemView(UUID id, String name, Map<String, String> data) {
+                    viewsActual.add(name);
+                }
+            });
+
+            assertEquals(viewsExpected, viewsActual);
+        }
+    }
+}