W 15397829.carbonj read write pool

carbonj.service/Dockerfile

@@ -5,15 +5,18 @@
 # For full license text, see the LICENSE file in the repo root or https://opensource.org/licenses/BSD-3-Clause
 #
 
-FROM quay.io/centos/centos:stream8
+#FROM quay.io/centos/centos:stream8
+FROM centos:8
 ARG DEPENDENCY=build/docker/dependency
 ARG VERSION=latest
 
 # install dependencies
 # Install aiohttp required by the script
 # Ensure crontab starts
 # Add Tini for proper sigkill handling https://github.com/krallin/tini
-RUN yum update -y && \
+RUN sed -i 's/mirrorlist/#mirrorlist/g' /etc/yum.repos.d/CentOS-* && \
+  sed -i 's|#baseurl=http://mirror.centos.org|baseurl=http://vault.centos.org|g' /etc/yum.repos.d/CentOS-* && \
+  yum update -y && \
   yum install -y libevent perl wget util-linux-ng cronie procps nc lsof sysstat epel-release && \
   yum install -y gcc-c++ gcc make libtool automake autoconf make python3-devel && \
   rpm --import http://repos.azulsystems.com/RPM-GPG-KEY-azulsystems && \

carbonj.service/build.gradle

@@ -19,14 +19,27 @@ plugins {
     id 'maven-publish'
     id 'net.researchgate.release'
     id 'org.springframework.boot'
+    id 'com.google.protobuf' version "0.9.3"
 }
 
 license {
     header rootProject.file('LICENSE-HEADER-JAVA')
     exclude "**/banner.txt"
+    exclude "**/MetricsResponse.java"
 //    include "**/*.conf"
 }
 
+sourceSets {
+    main {
+        proto {
+            srcDir 'src/main/protobuf'
+        }
+        java {
+            srcDirs 'build/generated/source/proto/main/java'
+        }
+    }
+}
+
 task licenseFormatDockerFiles(type: com.hierynomus.gradle.license.tasks.LicenseFormat) {
     source = fileTree(dir: "src").include("main/docker/files/*")
 }
@@ -37,6 +50,16 @@ task licenseCheckDockerFiles(type: com.hierynomus.gradle.license.tasks.LicenseFo
 }
 licenseMain.dependsOn licenseCheckDockerFiles
 
+protobuf {
+    protoc {
+        artifact = 'com.google.protobuf:protoc:3.14.0'
+
+        if (org.gradle.internal.os.OperatingSystem.current().isMacOsX()) {
+            artifact = 'com.google.protobuf:protoc:3.14.0:osx-x86_64'
+        }
+    }
+}
+
 apply plugin: 'java'
 apply plugin: 'eclipse'
 apply plugin: 'idea'
@@ -197,7 +220,7 @@ dependencies {
     implementation group: 'org.apache.logging.log4j', name: 'log4j-to-slf4j', version: '2.23.1'
     implementation group: 'org.slf4j', name: 'slf4j-api'
     implementation group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'
-    implementation group: 'org.rocksdb', name: 'rocksdbjni', version: '6.29.5'
+    implementation group: 'org.rocksdb', name: 'rocksdbjni', version: '6.20.3'
     implementation group: 'com.google.code.gson', name: 'gson', version: '2.10.1'
     implementation group: 'com.google.code.externalsortinginjava', name: 'externalsortinginjava', version: '0.6.2'
     implementation group: 'com.google.guava', name: 'guava', version: '33.0.0-jre'
@@ -214,6 +237,8 @@ dependencies {
     implementation group: 'io.dropwizard.metrics', name: 'metrics-graphite', version: '4.2.25'
     implementation 'jakarta.servlet:jakarta.servlet-api:6.0.0'
     implementation 'org.eclipse.jetty:jetty-server'
+    implementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.14.0'
+    implementation group: 'org.msgpack', name: 'jackson-dataformat-msgpack', version: '0.9.6'
 //    implementation 'org.mockito:mockito-core:4.11.0'
 
     testImplementation group: 'junit', name: 'junit', version: '4.13.2'

carbonj.service/src/main/java/com/demandware/carbonj/service/db/model/MsgPackMetric.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright (c) 2018, salesforce.com, inc.
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ * For full license text, see the LICENSE file in the repo root or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.demandware.carbonj.service.db.model;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import com.demandware.carbonj.service.accumulator.MetricAggregationPolicy;
+import com.google.common.base.Preconditions;
+
+import com.demandware.carbonj.service.strings.StringsCache;
+
+public class MsgPackMetric
+{
+    final public String path;
+    final public boolean isLeaf;
+
+    public MsgPackMetric( Metric metric )
+    {
+        this.path = metric.name;
+        this.isLeaf = metric.isLeaf();
+    }
+}

carbonj.service/src/main/java/com/demandware/carbonj/service/db/model/MsgPackSeries.java

@@ -0,0 +1,69 @@
+/**
+ * Copyright (c) 2018, salesforce.com, inc.
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ * For full license text, see the LICENSE file in the repo root or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.demandware.carbonj.service.db.model;
+
+import java.util.List;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonCreator;
+
+
+/**
+ * Represents series data to be sent to graphite.
+ */
+public class MsgPackSeries
+{
+    @JsonProperty("start")
+    final public long start;
+    @JsonProperty("end")
+    final public long end;
+    @JsonProperty("step")
+    final public long step;
+    @JsonProperty("name")
+    final public String name;
+    @JsonProperty("pathExpression")
+    final public String pathExpression;
+    @JsonProperty("values")
+    final public List<Double> values;
+
+    public MsgPackSeries( Series series)
+    {
+        this.start = series.start;
+        this.end = series.end;
+        this.step = series.step;
+        this.name = series.name;
+        this.pathExpression = series.name;
+        this.values = series.values;
+    }
+
+    @JsonCreator
+    public MsgPackSeries( @JsonProperty( "start" ) long start, @JsonProperty( "end" ) long end,
+                          @JsonProperty( "step" ) long step, @JsonProperty( "name" ) String name,
+                          @JsonProperty( "pathExpression" ) String pathExpression,
+                          @JsonProperty( "values" ) List<Double> values )
+    {
+        this.start = start;
+        this.end = end;
+        this.step = step;
+        this.name = name;
+        this.pathExpression = pathExpression;
+        this.values = values;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "MsgPackSeries{" +
+                "start=" + start +
+                ", end=" + end +
+                ", step=" + step +
+                ", name=" + name +
+                ", pathExpression=" + pathExpression +
+                ", values=" + values +
+                '}';
+    }
+}

carbonj.service/src/main/java/com/demandware/carbonj/service/db/points/DataPointArchiveRocksDB.java

@@ -130,20 +130,16 @@ class DataPointArchiveRocksDB
         this.catchUpTimer = metricRegistry.timer(MetricUtils.dbCatchUpTimerName(dbName));
         this.catchUpTimerError = metricRegistry.meter(MetricUtils.dbCatchUpTimerErrorName(dbName));
         this.longId = longId;
-        if (rocksdbConfig.readOnly) {
-            this.cleaner = null;
-        } else {
-            this.cleaner = new ThreadPoolExecutor( 1, 1, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(
-                    100000 ), new ThreadFactoryBuilder().setDaemon( true ).build(), new ThreadPoolExecutor.DiscardPolicy()
+        this.cleaner = new ThreadPoolExecutor( 1, 1, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(
+                100000 ), new ThreadFactoryBuilder().setDaemon( true ).build(), new ThreadPoolExecutor.DiscardPolicy()
+        {
+            @Override
+            public void rejectedExecution( Runnable r, ThreadPoolExecutor e )
             {
-                @Override
-                public void rejectedExecution( Runnable r, ThreadPoolExecutor e )
-                {
-                    log.info( "cleaner queue is full. rejecting (GC should pick this object up eventually)" );
-                    super.rejectedExecution( r, e );
-                }
-            } );
-        }
+                log.info( "cleaner queue is full. rejecting (GC should pick this object up eventually)" );
+                super.rejectedExecution( r, e );
+            }
+        } );
         TtlDB.loadLibrary();
     }
 
@@ -207,9 +203,7 @@ public void deleteMetric( long metricId, int from, int until )
             {
                 final RocksIterator iterToDispose = iter;
                 // contains global lock. Dispose in a separate thread to avoid contention.
-                if (cleaner != null) {
-                    cleaner.execute( ( ) -> dispose( iterToDispose ) );
-                }
+                cleaner.execute( ( ) -> dispose( iterToDispose ) );
             }
         }
     }
@@ -258,9 +252,7 @@ public long delete( int ts )
             {
                 final RocksIterator iterToDispose = iter;
                 // contains global lock. Dispose in a separate thread to avoid contention.
-                if (cleaner != null) {
-                    cleaner.execute(() -> dispose(iterToDispose));
-                }
+                cleaner.execute(() -> dispose(iterToDispose));
             }
         }
 
@@ -427,9 +419,7 @@ private void dispose( RocksObject o )
         if ( o != null )
         {
             // contains global lock. Dispose in a separate thread to avoid contention.
-            if (cleaner != null) {
-                cleaner.execute(o::close);
-            }
+            cleaner.execute(o::close);
         }
     }
 
@@ -495,9 +485,7 @@ public List<Double> getDataPoints( long metricId, int startTime, int endTime, in
             {
                 final RocksIterator iterToDispose = iter;
                 // contains global lock. Dispose in a separate thread to avoid contention.
-                if (cleaner != null) {
-                    cleaner.execute(() -> dispose(iterToDispose));
-                }
+                cleaner.execute(() -> dispose(iterToDispose));
             }
         }
 

carbonj.service/src/main/java/com/demandware/carbonj/service/engine/GraphiteMetricSearchServlet.java

@@ -7,9 +7,12 @@
 package com.demandware.carbonj.service.engine;
 
 import java.io.IOException;
+import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.List;
 
+import com.demandware.carbonj.service.db.model.MsgPackMetric;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
 import jakarta.servlet.ServletConfig;
 import jakarta.servlet.ServletException;
@@ -20,7 +23,9 @@
 import com.demandware.carbonj.service.events.EventsLogger;
 import com.demandware.carbonj.service.db.TimeSeriesStore;
 import com.demandware.carbonj.service.db.model.Metric;
+import com.demandware.carbonj.service.engine.protobuf.MetricsResponse;
 import com.google.gson.Gson;
+import org.msgpack.jackson.dataformat.MessagePackFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -81,11 +86,24 @@ protected void doGet( HttpServletRequest req, HttpServletResponse res )
         Preconditions.checkNotNull(query);
         boolean randomTest = req.getParameter("randomTest") != null;
 
+        boolean protobuf = "protobuf".equals( format );
+        boolean msgpack = "msgpack".equals( format );
         boolean json = "json".equals(format);
         if( json )
         {
             res.setContentType( "application/json" );
         }
+        else if ( protobuf )
+        {
+            LOG.info( "carbonapi request: found protobuf request" );
+            res.setContentType( "application/protobuf" );
+            //LOG.info( "carbonapi request: query: " + query + " --- blacklist: " + queryBlacklist );
+        }
+        else if ( msgpack )
+        {
+            LOG.info( "carbonapi request: found msgpack request" );
+            res.setContentType("application/octet-stream");
+        }
         else
         {
             res.setContentType( "application/pickle" );
@@ -120,6 +138,64 @@ protected void doGet( HttpServletRequest req, HttpServletResponse res )
             res.getWriter().write( gson.toJson( metrics ) );
             res.getWriter().close();
         }
+        else if ( msgpack )
+        {
+            ObjectMapper objectMapper = new ObjectMapper( new MessagePackFactory() );
+
+            List<MsgPackMetric> msgPackMetrics = new ArrayList<MsgPackMetric>();
+
+            for ( Metric metric : metrics )
+            {
+                msgPackMetrics.add( new MsgPackMetric( metric ) );
+            }
+
+            OutputStream output = res.getOutputStream();
+            try
+            {
+                // Serialize the metrics
+                byte[] serialized = objectMapper.writeValueAsBytes( msgPackMetrics );
+                res.setContentLength( serialized.length );
+                output.write( serialized );
+
+            }
+            catch ( IOException e )
+            {
+                LOG.error( "carbonapi request: error writing msgpack response", e.getMessage() );
+            }
+            finally
+            {
+                output.close();
+            }
+        }
+        else if (protobuf) {
+            //LOG.info( "carbonapi request: formatting response" );
+            OutputStream output = res.getOutputStream();
+
+            List<MetricsResponse.Metric> metricList = new ArrayList<MetricsResponse.Metric>();
+            for ( Metric metric : metrics )
+            {
+                MetricsResponse.Metric metricResult = MetricsResponse.Metric.newBuilder().setName(metric.name).setIsLeaf(metric.isLeaf()).build();
+                metricList.add(metricResult);
+            }
+
+            MetricsResponse.MetricList response =
+                    MetricsResponse.MetricList.newBuilder().setName("FindResponse").addAllMetrics(metricList).build();
+
+            //LOG.info( "carbonapi request: done formatting response " + response);
+            try
+            {
+                //LOG.info( "carbonapi request: writing response" );
+                response.writeTo( output );
+            }
+            catch ( Exception e )
+            {
+                LOG.error( "carbonapi request: error writing response", e.getMessage() );
+            }
+            finally
+            {
+                output.close();
+            }
+        }
         else
         {
             new GraphitePickler().pickleMetrics( metrics, res.getOutputStream() );