diff options
83 files changed, 1342 insertions, 319 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt index eba69f6ed02..379b6c5830f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -27,6 +27,7 @@ include_directories(BEFORE ${CMAKE_BINARY_DIR}/configdefinitions/src) add_subdirectory(application-model) add_subdirectory(application-preprocessor) +add_subdirectory(athenz-identity-provider-service) add_subdirectory(chain) add_subdirectory(component) add_subdirectory(config-bundle) diff --git a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java index 47160bdd4a2..d0ec98e3297 100644 --- a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java +++ b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java @@ -54,8 +54,10 @@ public interface ModelContext { boolean isFirstTimeDeployment(); boolean useDedicatedNodeForLogserver(); boolean useFdispatchByDefault(); + boolean dispatchWithProtobuf(); boolean useAdaptiveDispatch(); - boolean useSeparateServiceTypeForLogserverContainer(); + // TODO: Remove when 7.33 is the oldest model in use + default boolean useSeparateServiceTypeForLogserverContainer() { return true; } boolean enableMetricsProxyContainer(); } diff --git a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java index 89433cdf7c0..fe6d683adf8 100644 --- a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java +++ b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java @@ -35,8 +35,8 @@ public class TestProperties implements ModelContext.Properties { private boolean isFirstTimeDeployment = false; private boolean useDedicatedNodeForLogserver = false; private boolean useFdispatchByDefault = true; + private boolean dispatchWithProtobuf = true; private boolean useAdaptiveDispatch = false; - private boolean useSeparateServiceTypeForLogserverContainer = false; private boolean enableMetricsProxyContainer = false; @@ -54,7 +54,7 @@ public class TestProperties implements ModelContext.Properties { @Override public boolean useAdaptiveDispatch() { return useAdaptiveDispatch; } @Override public boolean useDedicatedNodeForLogserver() { return useDedicatedNodeForLogserver; } @Override public boolean useFdispatchByDefault() { return useFdispatchByDefault; } - @Override public boolean useSeparateServiceTypeForLogserverContainer() { return useSeparateServiceTypeForLogserverContainer; } + @Override public boolean dispatchWithProtobuf() { return dispatchWithProtobuf; } @Override public boolean enableMetricsProxyContainer() { return enableMetricsProxyContainer; } public TestProperties setApplicationId(ApplicationId applicationId) { @@ -87,11 +87,6 @@ public class TestProperties implements ModelContext.Properties { return this; } - public TestProperties setUseSeparateServiceTypeForLogserverContainer(boolean useSeparateServiceTypeForLogserverContainer) { - this.useSeparateServiceTypeForLogserverContainer = useSeparateServiceTypeForLogserverContainer; - return this; - } - public TestProperties setEnableMetricsProxyContainer(boolean enableMetricsProxyContainer) { this.enableMetricsProxyContainer = enableMetricsProxyContainer; return this; diff --git a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainer.java b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainer.java index 3d8773dcc2a..09bc29b181a 100644 --- a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainer.java +++ b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainer.java @@ -5,6 +5,7 @@ import com.yahoo.config.model.producer.AbstractConfigProducer; import com.yahoo.vespa.model.container.Container; import com.yahoo.vespa.model.container.ContainerCluster; import com.yahoo.config.model.api.container.ContainerServiceType; +import com.yahoo.vespa.model.container.component.AccessLogComponent; import com.yahoo.vespa.model.container.component.Handler; /** @@ -18,6 +19,7 @@ public class LogserverContainer extends Container { public LogserverContainer(AbstractConfigProducer parent, boolean useSeparateServiceTypeForLogserverContainer) { super(parent, "" + 0, 0); this.useSeparateServiceTypeForLogserverContainer = useSeparateServiceTypeForLogserverContainer; + addComponent(new AccessLogComponent(AccessLogComponent.AccessLogType.jsonAccessLog, ((LogserverContainerCluster) parent).getName(), true)); } @Override @@ -25,5 +27,4 @@ public class LogserverContainer extends Container { return useSeparateServiceTypeForLogserverContainer ? ContainerServiceType.LOGSERVER_CONTAINER : ContainerServiceType.CONTAINER; } - } diff --git a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java index f8f515cb609..184b75ceb2e 100644 --- a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java +++ b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java @@ -16,6 +16,7 @@ public class LogserverContainerCluster extends ContainerCluster<LogserverContain addDefaultHandlersWithVip(); addLogHandler(); + addDefaultSearchAccessLog(); } @Override diff --git a/config-model/src/main/java/com/yahoo/vespa/model/search/IndexedSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/search/IndexedSearchCluster.java index 9caf7fbdc9e..ee82d0cd719 100644 --- a/config-model/src/main/java/com/yahoo/vespa/model/search/IndexedSearchCluster.java +++ b/config-model/src/main/java/com/yahoo/vespa/model/search/IndexedSearchCluster.java @@ -107,6 +107,7 @@ public class IndexedSearchCluster extends SearchCluster private final DispatchGroup rootDispatch; private DispatchSpec dispatchSpec; private final boolean useFdispatchByDefault; + private final boolean dispatchWithProtobuf; private final boolean useAdaptiveDispatch; private List<SearchNode> searchNodes = new ArrayList<>(); @@ -126,6 +127,7 @@ public class IndexedSearchCluster extends SearchCluster dispatchParent = new SimpleConfigProducer(this, "dispatchers"); rootDispatch = new DispatchGroup(this); useFdispatchByDefault = deployState.getProperties().useFdispatchByDefault(); + dispatchWithProtobuf = deployState.getProperties().dispatchWithProtobuf(); useAdaptiveDispatch = deployState.getProperties().useAdaptiveDispatch(); } @@ -437,6 +439,7 @@ public class IndexedSearchCluster extends SearchCluster builder.maxNodesDownPerGroup(rootDispatch.getMaxNodesDownPerFixedRow()); builder.useMultilevelDispatch(useMultilevelDispatchSetup()); builder.useFdispatchByDefault(useFdispatchByDefault); + builder.dispatchWithProtobuf(dispatchWithProtobuf); builder.useLocalNode(tuning.dispatch.useLocalNode); builder.searchableCopies(rootDispatch.getSearchableCopies()); if (searchCoverage != null) { diff --git a/config-model/src/main/javacc/SDParser.jj b/config-model/src/main/javacc/SDParser.jj index 291532ff958..de5146ec7d2 100644 --- a/config-model/src/main/javacc/SDParser.jj +++ b/config-model/src/main/javacc/SDParser.jj @@ -570,6 +570,7 @@ void bodycfg(SDDocumentType document) : { } */ void compression(SDDocumentType document, String name) : { + deployLogger.log(Level.WARNING, "'compression' for a document is deprecated and ignored"); CompressionConfig cfg = new CompressionConfig(CompressionType.LZ4); } { diff --git a/config-model/src/test/java/com/yahoo/config/model/provision/ModelProvisioningTest.java b/config-model/src/test/java/com/yahoo/config/model/provision/ModelProvisioningTest.java index f395ef680da..6c9b9fdc084 100644 --- a/config-model/src/test/java/com/yahoo/config/model/provision/ModelProvisioningTest.java +++ b/config-model/src/test/java/com/yahoo/config/model/provision/ModelProvisioningTest.java @@ -956,27 +956,7 @@ public class ModelProvisioningTest { " </container>" + "</services>"; boolean useDedicatedNodeForLogserver = false; - boolean useSeparateServiceTypeForLogserverContainer = false; - testContainerOnLogserverHost(services, useDedicatedNodeForLogserver, useSeparateServiceTypeForLogserverContainer); - } - - @Test - public void testLogserverContainerWhenDedicatedLogserverSeparateServiceType() { - String services = - "<?xml version='1.0' encoding='utf-8' ?>\n" + - "<services>" + - " <admin version='4.0'>" + - " <logservers>" + - " <nodes count='1' dedicated='true'/>" + - " </logservers>" + - " </admin>" + - " <container version='1.0' id='foo'>" + - " <nodes count='1'/>" + - " </container>" + - "</services>"; - boolean useDedicatedNodeForLogserver = false; - boolean useSeparateServiceTypeForLogserverContainer = true; - testContainerOnLogserverHost(services, useDedicatedNodeForLogserver, useSeparateServiceTypeForLogserverContainer); + testContainerOnLogserverHost(services, useDedicatedNodeForLogserver); } @Test @@ -989,8 +969,7 @@ public class ModelProvisioningTest { " </container>" + "</services>"; boolean useDedicatedNodeForLogserver = true; - boolean useSeparateServiceTypeForLogserverContainer = false; - testContainerOnLogserverHost(services, useDedicatedNodeForLogserver, useSeparateServiceTypeForLogserverContainer); + testContainerOnLogserverHost(services, useDedicatedNodeForLogserver); } @Test @@ -1833,11 +1812,10 @@ public class ModelProvisioningTest { // Tests that a container is allocated on logserver host and that // it is able to get config - private void testContainerOnLogserverHost(String services, boolean useDedicatedNodeForLogserver, boolean useSeparateServiceTypeForLogserverContainer) { + private void testContainerOnLogserverHost(String services, boolean useDedicatedNodeForLogserver) { int numberOfHosts = 2; VespaModelTester tester = new VespaModelTester(); tester.useDedicatedNodeForLogserver(useDedicatedNodeForLogserver); - tester.useSeparateServiceTypeForLogserverContainer(useSeparateServiceTypeForLogserverContainer); tester.addHosts(numberOfHosts); VespaModel model = tester.createModel(Zone.defaultZone(), services, true); @@ -1847,9 +1825,7 @@ public class ModelProvisioningTest { Logserver logserver = admin.getLogserver(); HostResource hostResource = logserver.getHostResource(); assertNotNull(hostResource.getService("logserver")); - String containerServiceType = useSeparateServiceTypeForLogserverContainer - ? ContainerServiceType.LOGSERVER_CONTAINER.serviceName - : ContainerServiceType.CONTAINER.serviceName; + String containerServiceType = ContainerServiceType.LOGSERVER_CONTAINER.serviceName; assertNotNull(hostResource.getService(containerServiceType)); // Test that the container gets config diff --git a/config-model/src/test/java/com/yahoo/vespa/model/admin/DedicatedAdminV4Test.java b/config-model/src/test/java/com/yahoo/vespa/model/admin/DedicatedAdminV4Test.java index f004cc8ff2a..3896b6d799a 100644 --- a/config-model/src/test/java/com/yahoo/vespa/model/admin/DedicatedAdminV4Test.java +++ b/config-model/src/test/java/com/yahoo/vespa/model/admin/DedicatedAdminV4Test.java @@ -28,6 +28,7 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; import static com.yahoo.config.model.api.container.ContainerServiceType.CONTAINER; +import static com.yahoo.config.model.api.container.ContainerServiceType.LOGSERVER_CONTAINER; import static com.yahoo.config.model.api.container.ContainerServiceType.METRICS_PROXY_CONTAINER; import static com.yahoo.config.model.api.container.ContainerServiceType.QRSERVER; import static org.junit.Assert.assertEquals; @@ -77,7 +78,7 @@ public class DedicatedAdminV4Test { assertHostContainsServices(model, "hosts/myhost0", "slobrok", "logd"); assertHostContainsServices(model, "hosts/myhost1", "slobrok", "logd"); // Note: A container is always added on logserver host - assertHostContainsServices(model, "hosts/myhost2", "logserver", "logd", CONTAINER.serviceName); + assertHostContainsServices(model, "hosts/myhost2", "logserver", "logd", LOGSERVER_CONTAINER.serviceName); Monitoring monitoring = model.getAdmin().getMonitoring(); assertEquals("vespa.routing", monitoring.getClustername()); @@ -157,7 +158,7 @@ public class DedicatedAdminV4Test { assertHostContainsServices(model, "hosts/myhost0", "logd", "logforwarder", "slobrok"); assertHostContainsServices(model, "hosts/myhost1", "logd", "logforwarder", "slobrok"); // Note: A container is always added on logserver host - assertHostContainsServices(model, "hosts/myhost2", "logd", "logforwarder", "logserver", CONTAINER.serviceName); + assertHostContainsServices(model, "hosts/myhost2", "logd", "logforwarder", "logserver", LOGSERVER_CONTAINER.serviceName); Set<String> configIds = model.getConfigIds(); // 1 logforwarder on each host @@ -200,7 +201,7 @@ public class DedicatedAdminV4Test { .properties(new TestProperties().setHostedVespa(true))); assertEquals(1, model.getHosts().size()); // Should create a container on the same node as logserver - assertHostContainsServices(model, "hosts/myhost0", "slobrok", "logd", "logserver", CONTAINER.serviceName); + assertHostContainsServices(model, "hosts/myhost0", "slobrok", "logd", "logserver", LOGSERVER_CONTAINER.serviceName); } private Set<String> serviceNames(VespaModel model, String hostname) { diff --git a/config-model/src/test/java/com/yahoo/vespa/model/test/VespaModelTester.java b/config-model/src/test/java/com/yahoo/vespa/model/test/VespaModelTester.java index 00bc82bbe02..801e138f3c7 100644 --- a/config-model/src/test/java/com/yahoo/vespa/model/test/VespaModelTester.java +++ b/config-model/src/test/java/com/yahoo/vespa/model/test/VespaModelTester.java @@ -47,7 +47,6 @@ public class VespaModelTester { private Map<String, Collection<Host>> hostsByFlavor = new HashMap<>(); private ApplicationId applicationId = ApplicationId.defaultId(); private boolean useDedicatedNodeForLogserver = false; - private boolean useSeparateServiceTypeForLogserverContainer = false; private boolean enableMetricsProxyContainer = false; public VespaModelTester() { @@ -99,10 +98,6 @@ public class VespaModelTester { this.useDedicatedNodeForLogserver = useDedicatedNodeForLogserver; } - public void useSeparateServiceTypeForLogserverContainer(boolean useSeparateServiceTypeForLogserverContainer) { - this.useSeparateServiceTypeForLogserverContainer = useSeparateServiceTypeForLogserverContainer; - } - public void enableMetricsProxyContainer(boolean enableMetricsProxyContainer) { this.enableMetricsProxyContainer = enableMetricsProxyContainer; } @@ -148,7 +143,6 @@ public class VespaModelTester { .setHostedVespa(hosted) .setApplicationId(applicationId) .setUseDedicatedNodeForLogserver(useDedicatedNodeForLogserver) - .setUseSeparateServiceTypeForLogserverContainer(useSeparateServiceTypeForLogserverContainer) .setEnableMetricsProxyContainer(enableMetricsProxyContainer); DeployState deployState = new DeployState.Builder() diff --git a/configdefinitions/src/vespa/dispatch.def b/configdefinitions/src/vespa/dispatch.def index f0a4b0d8419..7d5979bcdf1 100644 --- a/configdefinitions/src/vespa/dispatch.def +++ b/configdefinitions/src/vespa/dispatch.def @@ -19,6 +19,9 @@ distributionPolicy enum { ROUNDROBIN, ADAPTIVE } default=ROUNDROBIN # Should fdispatch be used as the default dispatcher useFdispatchByDefault bool default=true +# Should protobuf/jrt be preferred over fs4 +dispatchWithProtobuf bool default=false + # Is multi-level dispatch configured for this cluster useMultilevelDispatch bool default=false diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java index 780135c893e..0279d175488 100644 --- a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java +++ b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java @@ -131,7 +131,7 @@ public class ModelContextImpl implements ModelContext { private final boolean useDedicatedNodeForLogserver; private final boolean useFdispatchByDefault; private final boolean useAdaptiveDispatch; - private final boolean useSeparateServiceTypeForLogserverContainer; + private final boolean dispatchWithProtobuf; private final boolean enableMetricsProxyContainer; public Properties(ApplicationId applicationId, @@ -161,9 +161,9 @@ public class ModelContextImpl implements ModelContext { .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value(); this.useFdispatchByDefault = Flags.USE_FDISPATCH_BY_DEFAULT.bindTo(flagSource) .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value(); - this.useAdaptiveDispatch = Flags.USE_ADAPTIVE_DISPATCH.bindTo(flagSource) + this.dispatchWithProtobuf = Flags.DISPATCH_WITH_PROTOBUF.bindTo(flagSource) .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value(); - this.useSeparateServiceTypeForLogserverContainer = Flags.USE_SEPARATE_SERVICE_TYPE_FOR_LOGSERVER_CONTAINER.bindTo(flagSource) + this.useAdaptiveDispatch = Flags.USE_ADAPTIVE_DISPATCH.bindTo(flagSource) .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value(); this.enableMetricsProxyContainer = Flags.ENABLE_METRICS_PROXY_CONTAINER.bindTo(flagSource) .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value(); @@ -213,10 +213,10 @@ public class ModelContextImpl implements ModelContext { public boolean useFdispatchByDefault() { return useFdispatchByDefault; } @Override - public boolean useAdaptiveDispatch() { return useAdaptiveDispatch; } + public boolean dispatchWithProtobuf() { return dispatchWithProtobuf; } @Override - public boolean useSeparateServiceTypeForLogserverContainer() { return useSeparateServiceTypeForLogserverContainer; } + public boolean useAdaptiveDispatch() { return useAdaptiveDispatch; } @Override public boolean enableMetricsProxyContainer() { return enableMetricsProxyContainer; } diff --git a/container-core/CMakeLists.txt b/container-core/CMakeLists.txt index 1a2bbabaed3..43225e38aee 100644 --- a/container-core/CMakeLists.txt +++ b/container-core/CMakeLists.txt @@ -12,3 +12,5 @@ install_config_definition(src/main/resources/configdefinitions/qr.def container. install_config_definition(src/main/resources/configdefinitions/servlet-config.def container.servlet.servlet-config.def) install_config_definition(src/main/resources/configdefinitions/threadpool.def container.handler.threadpool.def) install_config_definition(src/main/resources/configdefinitions/vip-status.def container.core.vip-status.def) + +vespa_install_script(src/main/sh/vespa-load-balancer-status libexec/vespa) diff --git a/container-core/src/main/java/com/yahoo/container/handler/LogReader.java b/container-core/src/main/java/com/yahoo/container/handler/LogReader.java index 95a0e9a6766..663741f9bef 100644 --- a/container-core/src/main/java/com/yahoo/container/handler/LogReader.java +++ b/container-core/src/main/java/com/yahoo/container/handler/LogReader.java @@ -57,13 +57,10 @@ class LogReader { void writeLogs(OutputStream outputStream, Instant earliestLogThreshold, Instant latestLogThreshold) { try { for (Path file : getMatchingFiles(earliestLogThreshold, latestLogThreshold)) { - if (file.toString().endsWith(".gz")) { - Files.copy(file, outputStream); - } else { - OutputStream zip = new GZIPOutputStream(outputStream); - Files.copy(file, zip); - zip.close(); + if (!file.toString().endsWith(".gz") && !(outputStream instanceof GZIPOutputStream)) { + outputStream = new GZIPOutputStream(outputStream); } + Files.copy(file, outputStream); } outputStream.close(); } catch (IOException e) { diff --git a/container-core/src/main/sh/vespa-load-balancer-status b/container-core/src/main/sh/vespa-load-balancer-status new file mode 100755 index 00000000000..e93337333f3 --- /dev/null +++ b/container-core/src/main/sh/vespa-load-balancer-status @@ -0,0 +1,214 @@ +#!/bin/bash +# +# Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +# BEGIN environment bootstrap section +# Do not edit between here and END as this section should stay identical in all scripts + +findpath () { + myname=${0} + mypath=${myname%/*} + myname=${myname##*/} + if [ "$mypath" ] && [ -d "$mypath" ]; then + return + fi + mypath=$(pwd) + if [ -f "${mypath}/${myname}" ]; then + return + fi + echo "FATAL: Could not figure out the path where $myname lives from $0" + exit 1 +} + +COMMON_ENV=libexec/vespa/common-env.sh + +source_common_env () { + if [ "$VESPA_HOME" ] && [ -d "$VESPA_HOME" ]; then + export VESPA_HOME + common_env=$VESPA_HOME/$COMMON_ENV + if [ -f "$common_env" ]; then + . $common_env + return + fi + fi + return 1 +} + +findroot () { + source_common_env && return + if [ "$VESPA_HOME" ]; then + echo "FATAL: bad VESPA_HOME value '$VESPA_HOME'" + exit 1 + fi + if [ "$ROOT" ] && [ -d "$ROOT" ]; then + VESPA_HOME="$ROOT" + source_common_env && return + fi + findpath + while [ "$mypath" ]; do + VESPA_HOME=${mypath} + source_common_env && return + mypath=${mypath%/*} + done + echo "FATAL: missing VESPA_HOME environment variable" + echo "Could not locate $COMMON_ENV anywhere" + exit 1 +} + +findhost () { + if [ "${VESPA_HOSTNAME}" = "" ]; then + VESPA_HOSTNAME=$(vespa-detect-hostname || hostname -f || hostname || echo "localhost") || exit 1 + fi + validate="${VESPA_HOME}/bin/vespa-validate-hostname" + if [ -f "$validate" ]; then + "$validate" "${VESPA_HOSTNAME}" || exit 1 + fi + export VESPA_HOSTNAME +} + +findroot +findhost + +# END environment bootstrap section + +set -eu + +declare LB_STATUS_DIR="$VESPA_HOME"/var/vespa/load-balancer +declare LB_STATUS_FILE="$LB_STATUS_DIR"/status.html +declare LB_OPERATOR_LOG="$LB_STATUS_DIR"/operator.log + +function Usage { + cat <<EOF +Usage: ${0##*/} COMMAND [-u USER] [-f] +Make jdisc container stop serving /status.html. + +Useful when jdisc container is behind a load balancer: The load balancer can be +set up to monitor the health of /status.html requests, and remove bad backends +from serving. + +Command: + get Return info on the current in/out status. + in Undo 'out'. This command is a no-op if 1. status is already in, or 2. + if the the user that set it out is different from USER and -f was NOT + specified. + out Stop answering OK on /status.html requests against jdisc container. + Note: The jdisc container may not answer OK for other reasons too. + +Options: + -u USER Set the user agent. The user setting the status in, must match + the user that set it out. Defaults to current user. + -f Force-set status: Ignore any mismatch on user. +EOF + + exit 0 +} + +function PrintPair { + printf "%-19s %s\n" "$1:" "$2" +} + +function IsIn { + if [ -r "$LB_STATUS_FILE" ]; then + return 0 + else + return 1 + fi +} + +function DifferentUserSetOut { + local user="$1" + + if [ -r "$LB_OPERATOR_LOG" ]; then + local out_user + out_user=$(< "$LB_OPERATOR_LOG") + if [ "$user" != "$out_user" ]; then + return 0 + fi + fi + + return 1 +} + +function GetCommand { + if IsIn; then + PrintPair "VIP status" IN + else + PrintPair "VIP status" OUT + fi + PrintPair "Status file" "$LB_STATUS_FILE" + + if [ -r "$LB_OPERATOR_LOG" ]; then + PrintPair "Last modified" "$(stat -c %y "$LB_OPERATOR_LOG")" + PrintPair "Last modified by" "$(< "$LB_OPERATOR_LOG")" + fi +} + +function InCommand { + local user="$1" + local force="$2" + + if ! $force; then + if IsIn || DifferentUserSetOut "$user"; then + return + fi + fi + + mkdir -p "$LB_STATUS_DIR" + echo "$user" > "$LB_OPERATOR_LOG" + echo OK > "$LB_STATUS_FILE" +} + +function OutCommand { + local user="$1" + local force="$2" + + if ! $force && ! IsIn; then + return + fi + + mkdir -p "$LB_STATUS_DIR" + echo "$user" > "$LB_OPERATOR_LOG" + rm -f "$LB_STATUS_FILE" +} + +function Main { + if (($# == 0)); then + Usage + fi + + local command= + local user="${SUDO_USER:-${USER:-$(id -nu)}}" + local force=false + + # Supports placement of options both before and after command. + while (($# > 0)); do + case "$1" in + -f) + force=true + shift + ;; + -u) + user="$2" + shift 2 + ;; + -*) Usage "Unknown option '$1'" ;; + *) + case "$1" in + get) command="GetCommand" ;; + in) command="InCommand" ;; + out) command="OutCommand" ;; + *) Usage ;; + esac + shift + ;; + esac + done + + if [ -z "$command" ]; then + Usage + fi + + "$command" "$user" "$force" +} + +Main "$@" diff --git a/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java b/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java index 1bd68ad0c48..464d6f772eb 100644 --- a/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java +++ b/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java @@ -69,14 +69,6 @@ public class LogReaderTest { @Test public void testZippedStreaming() throws IOException { - // Add some more files - Files.setLastModifiedTime( - Files.write(logDirectory.resolve("log3.gz"), compress("Three\n")), - FileTime.from(Instant.ofEpochMilli(324))); - Files.setLastModifiedTime( - Files.write(logDirectory.resolve("log4"), "Four\n".getBytes()), - FileTime.from(Instant.ofEpochMilli(432))); - ByteArrayOutputStream zippedBaos = new ByteArrayOutputStream(); LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*")); logReader.writeLogs(zippedBaos, Instant.ofEpochMilli(21), Instant.now()); @@ -85,7 +77,7 @@ public class LogReaderTest { Scanner s = new Scanner(unzippedIs).useDelimiter("\\A"); String actual = s.hasNext() ? s.next() : ""; - String expected = "This is one log file\nThis is another log file\nThree\nFour\n"; + String expected = "This is one log file\nThis is another log file\n"; assertEquals(expected, actual); } diff --git a/container-search/src/main/java/com/yahoo/fs4/BasicPacket.java b/container-search/src/main/java/com/yahoo/fs4/BasicPacket.java index 6f87e45af25..f87721dc503 100644 --- a/container-search/src/main/java/com/yahoo/fs4/BasicPacket.java +++ b/container-search/src/main/java/com/yahoo/fs4/BasicPacket.java @@ -26,9 +26,7 @@ public abstract class BasicPacket { private static int DEFAULT_WRITE_BUFFER_SIZE = (10 * 1024); public static final int CODE_MASK = 0x00ff_ffff; // Reserve upper byte for flags. - protected byte[] encodedBody; - - protected ByteBuffer encodingBuffer; + private byte[] encodedBody; /** The length of this packet in bytes or -1 if not known */ protected int length = -1; @@ -199,7 +197,7 @@ public abstract class BasicPacket { throw new UnsupportedOperationException("Encoding of " + this + " is not implemented"); } - protected void setEncodedBody(ByteBuffer b, int start, int length) { + private void setEncodedBody(ByteBuffer b, int start, int length) { encodedBody = new byte[length]; b.position(start); b.get(encodedBody); @@ -222,18 +220,7 @@ public abstract class BasicPacket { * * If this packet does not use a channel ID, the ID will be ignored. */ - public final void allocateAndEncode(int channelId) { - allocateAndEncode(channelId, DEFAULT_WRITE_BUFFER_SIZE); - } - - private void allocateAndEncode(int channelId, int initialSize) { - if (encodingBuffer != null) { - patchChannelId(encodingBuffer, channelId); - return; - } - - int size = initialSize; - ByteBuffer buffer = ByteBuffer.allocate(size); + private ByteBuffer allocateAndEncode(int channelId, ByteBuffer buffer) { while (true) { try { if (hasChannelId()) { @@ -242,43 +229,25 @@ public abstract class BasicPacket { encode(buffer); } buffer.flip(); - encodingBuffer = buffer; break; } catch (BufferTooSmallException e) { - size *= 2; - buffer = ByteBuffer.allocate(size); + buffer = ByteBuffer.allocate(buffer.capacity()*2); } } + return buffer; } - // No channel ID for BasicPacket instances, so it's a NOP - protected void patchChannelId(ByteBuffer buf, int channelId) {} - /** * Return buffer containing the encoded form of this package and * remove internal reference to it. */ public final ByteBuffer grantEncodingBuffer(int channelId) { - if (encodingBuffer == null) { - allocateAndEncode(channelId); - } else { - patchChannelId(encodingBuffer, channelId); - } - ByteBuffer b = encodingBuffer; - encodingBuffer = null; - return b; + return allocateAndEncode(channelId, ByteBuffer.allocate(DEFAULT_WRITE_BUFFER_SIZE)); } - public final ByteBuffer grantEncodingBuffer(int channelId, int initialSize) { - if (encodingBuffer == null) { - allocateAndEncode(channelId, initialSize); - } else { - patchChannelId(encodingBuffer, channelId); - } - ByteBuffer b = encodingBuffer; - encodingBuffer = null; - return b; + public final ByteBuffer grantEncodingBuffer(int channelId, ByteBuffer buffer) { + return allocateAndEncode(channelId, buffer); } /** Returns the code of this package */ diff --git a/container-search/src/main/java/com/yahoo/fs4/Packet.java b/container-search/src/main/java/com/yahoo/fs4/Packet.java index 78d5083c25f..1e9deede59d 100644 --- a/container-search/src/main/java/com/yahoo/fs4/Packet.java +++ b/container-search/src/main/java/com/yahoo/fs4/Packet.java @@ -19,8 +19,6 @@ public abstract class Packet extends BasicPacket { */ protected int channel = -1; - private static final int CHANNEL_ID_OFFSET = 8; - /** * Fills this package from a byte buffer positioned at the first * byte of the package @@ -109,17 +107,6 @@ public abstract class Packet extends BasicPacket { return true; } - /** - * Only for use with encodingBuffer magic. - * - * This is only called from allocateAndEncode and grantEncodingBuffer, - * therefore an assumption about the packet starting at the beginning of the - * buffer is made. - */ - protected void patchChannelId(ByteBuffer buf, int channelId) { - buf.putInt(CHANNEL_ID_OFFSET, channelId); - } - public String toString() { return "packet with code " + getCode() + ", channelId=" + getChannel(); } diff --git a/container-search/src/main/java/com/yahoo/fs4/PongPacket.java b/container-search/src/main/java/com/yahoo/fs4/PongPacket.java index 13fb7d84408..37aaf7067a9 100644 --- a/container-search/src/main/java/com/yahoo/fs4/PongPacket.java +++ b/container-search/src/main/java/com/yahoo/fs4/PongPacket.java @@ -28,7 +28,7 @@ public class PongPacket extends BasicPacket { /** For testing */ public PongPacket(long activeDocuments) { - this.activeDocuments = Optional.of(activeDocuments); + this.activeDocuments = Optional.of(activeDocuments); } private int code; diff --git a/container-search/src/main/java/com/yahoo/fs4/mplex/Backend.java b/container-search/src/main/java/com/yahoo/fs4/mplex/Backend.java index 202ee94383f..12f8e9e387d 100644 --- a/container-search/src/main/java/com/yahoo/fs4/mplex/Backend.java +++ b/container-search/src/main/java/com/yahoo/fs4/mplex/Backend.java @@ -1,19 +1,29 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.fs4.mplex; - -import com.yahoo.fs4.*; +import com.yahoo.fs4.BasicPacket; +import com.yahoo.fs4.Packet; +import com.yahoo.fs4.PacketDumper; +import com.yahoo.fs4.PacketListener; +import com.yahoo.fs4.PacketNotificationsBroadcaster; +import com.yahoo.fs4.PacketQueryTracer; import com.yahoo.io.Connection; import com.yahoo.io.ConnectionFactory; import com.yahoo.io.Listener; import com.yahoo.vespa.defaults.Defaults; import com.yahoo.yolean.Exceptions; +import com.yahoo.yolean.concurrent.ConcurrentResourcePool; +import com.yahoo.yolean.concurrent.ResourceFactory; +import com.yahoo.yolean.concurrent.ResourcePool; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.nio.ByteBuffer; import java.nio.channels.SocketChannel; -import java.util.*; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Level; import java.util.logging.Logger; @@ -23,6 +33,8 @@ import java.util.logging.Logger; */ public class Backend implements ConnectionFactory { + private static int DEFAULT_BUFFER_SIZE = 0x8000; + public static final class BackendStatistics { public final int activeConnections; @@ -61,6 +73,12 @@ public class Backend implements ConnectionFactory { private final ConnectionPool connectionPool; private final PacketDumper packetDumper; private final AtomicInteger connectionCount = new AtomicInteger(0); + private final ConcurrentResourcePool<ByteBuffer> byteBufferResourcePool = new ConcurrentResourcePool<>(new ResourceFactory<>() { + @Override + public ByteBuffer create() { + return ByteBuffer.allocate(DEFAULT_BUFFER_SIZE); + } + }); /** * For unit testing. do not use @@ -116,6 +134,10 @@ public class Backend implements ConnectionFactory { return connection; } + ConcurrentResourcePool<ByteBuffer> getBufferPool() { + return byteBufferResourcePool; + } + /** * Return a connection to the connection pool. If the * connection is not valid anymore we drop it, ie. do not @@ -416,10 +438,6 @@ public class Backend implements ConnectionFactory { } } - public void dumpPackets(final PacketDumper.PacketType packetType, final boolean on) throws IOException { - packetDumper.dumpPackets(packetType, on); - } - public String getHost() { return host; } diff --git a/container-search/src/main/java/com/yahoo/fs4/mplex/FS4Connection.java b/container-search/src/main/java/com/yahoo/fs4/mplex/FS4Connection.java index 69267f4a6b2..7dcbefde9fa 100644 --- a/container-search/src/main/java/com/yahoo/fs4/mplex/FS4Connection.java +++ b/container-search/src/main/java/com/yahoo/fs4/mplex/FS4Connection.java @@ -36,7 +36,6 @@ public class FS4Connection implements Connection private static int idCounter = 1; private int idNumber; - private int maxInitialSize = 1024; // outbound data private ByteBuffer writeBuffer; @@ -69,7 +68,7 @@ public class FS4Connection implements Connection * Packet sending interface. */ public void sendPacket (BasicPacket packet, Integer channelId) throws IOException { - ByteBuffer buffer = packet.grantEncodingBuffer(channelId.intValue(), maxInitialSize); + ByteBuffer buffer = packet.grantEncodingBuffer(channelId.intValue(), backend.getBufferPool().alloc()); ByteBuffer viewForPacketListener = buffer.slice(); synchronized (this) { if (!(valid && channel.isOpen())) { @@ -79,9 +78,6 @@ public class FS4Connection implements Connection ", isOpen = " + channel.isOpen()); } - if (buffer.capacity() > maxInitialSize) { - maxInitialSize = buffer.limit(); - } if (writeBuffer == null) { writeBuffer = buffer; } else { @@ -131,6 +127,8 @@ public class FS4Connection implements Connection // buffer drained so we forget it and see what happens when we // go around. if indeed we go around if (!writeBuffer.hasRemaining()) { + writeBuffer.clear(); + backend.getBufferPool().free(writeBuffer); writeBuffer = null; } } while (bytesWritten > 0); diff --git a/container-search/src/main/java/com/yahoo/prelude/Pong.java b/container-search/src/main/java/com/yahoo/prelude/Pong.java index cef64c293af..a6bc3e7975d 100644 --- a/container-search/src/main/java/com/yahoo/prelude/Pong.java +++ b/container-search/src/main/java/com/yahoo/prelude/Pong.java @@ -1,15 +1,15 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.prelude; +import com.yahoo.fs4.PongPacket; +import com.yahoo.search.result.ErrorMessage; +import com.yahoo.search.statistics.ElapsedTime; + import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Optional; -import com.yahoo.fs4.PongPacket; -import com.yahoo.search.result.ErrorMessage; -import com.yahoo.search.statistics.ElapsedTime; - /** * An answer from Ping. * @@ -21,20 +21,29 @@ public class Pong { private final List<ErrorMessage> errors = new ArrayList<>(1); private final Optional<PongPacket> pongPacket; private ElapsedTime elapsed = new ElapsedTime(); + private final Optional<Long> activeDocuments; public Pong() { this.pongPacket = Optional.empty(); + this.activeDocuments = Optional.empty(); } - + public Pong(ErrorMessage error) { errors.add(error); this.pongPacket = Optional.empty(); + this.activeDocuments = Optional.empty(); } - + public Pong(PongPacket pongPacket) { this.pongPacket = Optional.of(pongPacket); + this.activeDocuments = Optional.empty(); } - + + public Pong(long activeDocuments) { + this.pongPacket = Optional.empty(); + this.activeDocuments = Optional.of(activeDocuments); + } + public void addError(ErrorMessage error) { errors.add(error); } @@ -49,6 +58,7 @@ public class Pong { /** Returns the number of active documents in the backend responding in this Pong, if available */ public Optional<Long> activeDocuments() { + if (activeDocuments.isPresent()) return activeDocuments; if ( ! pongPacket.isPresent()) return Optional.empty(); return pongPacket.get().getActiveDocuments(); } diff --git a/container-search/src/main/java/com/yahoo/prelude/fastsearch/FS4InvokerFactory.java b/container-search/src/main/java/com/yahoo/prelude/fastsearch/FS4InvokerFactory.java index ac99675c9c5..b9af60089f8 100644 --- a/container-search/src/main/java/com/yahoo/prelude/fastsearch/FS4InvokerFactory.java +++ b/container-search/src/main/java/com/yahoo/prelude/fastsearch/FS4InvokerFactory.java @@ -3,13 +3,16 @@ package com.yahoo.prelude.fastsearch; import com.google.common.collect.ImmutableMap; import com.yahoo.fs4.mplex.Backend; +import com.yahoo.prelude.Pong; import com.yahoo.search.Query; import com.yahoo.search.Result; +import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.dispatch.FillInvoker; import com.yahoo.search.dispatch.InterleavedFillInvoker; import com.yahoo.search.dispatch.InvokerFactory; import com.yahoo.search.dispatch.SearchInvoker; import com.yahoo.search.dispatch.searchcluster.Node; +import com.yahoo.search.dispatch.searchcluster.Pinger; import com.yahoo.search.dispatch.searchcluster.SearchCluster; import com.yahoo.search.result.Hit; @@ -20,6 +23,7 @@ import java.util.Iterator; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.Callable; /** * FS4InvokerFactory constructs {@link FillInvoker} and {@link SearchInvoker} objects that communicate with @@ -97,4 +101,8 @@ public class FS4InvokerFactory extends InvokerFactory { return requiredNodes; } + @Override + public Callable<Pong> createPinger(Node node, ClusterMonitor<Node> monitor) { + return new Pinger(node, monitor, fs4ResourcePool); + } } diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java index 74d9c38b273..f2dbb1e8557 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java @@ -56,6 +56,7 @@ public class Dispatcher extends AbstractComponent { private final LoadBalancer loadBalancer; private final boolean multilevelDispatch; private final boolean internalDispatchByDefault; + private final boolean dispatchWithProtobuf; private final FS4InvokerFactory fs4InvokerFactory; private final RpcInvokerFactory rpcInvokerFactory; @@ -65,14 +66,12 @@ public class Dispatcher extends AbstractComponent { public Dispatcher(String clusterId, DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, int containerClusterSize, VipStatus vipStatus, Metric metric) { - this(new SearchCluster(clusterId, dispatchConfig, fs4ResourcePool, containerClusterSize, vipStatus), dispatchConfig, - fs4ResourcePool, new RpcResourcePool(dispatchConfig), metric); + this(new SearchCluster(clusterId, dispatchConfig, containerClusterSize, vipStatus), dispatchConfig, fs4ResourcePool, metric); } - public Dispatcher(SearchCluster searchCluster, DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, - RpcResourcePool rpcResourcePool, Metric metric) { + public Dispatcher(SearchCluster searchCluster, DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, Metric metric) { this(searchCluster, dispatchConfig, new FS4InvokerFactory(fs4ResourcePool, searchCluster), - new RpcInvokerFactory(rpcResourcePool, searchCluster), metric); + new RpcInvokerFactory(new RpcResourcePool(dispatchConfig), searchCluster, dispatchConfig.dispatchWithProtobuf()), metric); } public Dispatcher(SearchCluster searchCluster, DispatchConfig dispatchConfig, FS4InvokerFactory fs4InvokerFactory, @@ -82,12 +81,15 @@ public class Dispatcher extends AbstractComponent { dispatchConfig.distributionPolicy() == DispatchConfig.DistributionPolicy.ROUNDROBIN); this.multilevelDispatch = dispatchConfig.useMultilevelDispatch(); this.internalDispatchByDefault = !dispatchConfig.useFdispatchByDefault(); + this.dispatchWithProtobuf = dispatchConfig.dispatchWithProtobuf(); this.fs4InvokerFactory = fs4InvokerFactory; this.rpcInvokerFactory = rpcInvokerFactory; this.metric = metric; this.metricContext = metric.createContext(null); + + searchCluster.startClusterMonitoring(dispatchWithProtobuf ? rpcInvokerFactory : fs4InvokerFactory); } /** Returns the search cluster this dispatches to */ @@ -120,7 +122,8 @@ public class Dispatcher extends AbstractComponent { return Optional.empty(); } - InvokerFactory factory = query.properties().getBoolean(dispatchProtobuf, false) ? rpcInvokerFactory : fs4InvokerFactory; + InvokerFactory factory = query.properties().getBoolean(dispatchProtobuf, dispatchWithProtobuf) + ? rpcInvokerFactory : fs4InvokerFactory; Optional<SearchInvoker> invoker = getSearchPathInvoker(query, factory, searcher); diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/InvokerFactory.java b/container-search/src/main/java/com/yahoo/search/dispatch/InvokerFactory.java index 815a2a257ea..8617c74ec41 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/InvokerFactory.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/InvokerFactory.java @@ -1,9 +1,11 @@ // Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.search.dispatch; +import com.yahoo.prelude.Pong; import com.yahoo.prelude.fastsearch.VespaBackEndSearcher; import com.yahoo.search.Query; import com.yahoo.search.Result; +import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.dispatch.searchcluster.Node; import com.yahoo.search.dispatch.searchcluster.SearchCluster; import com.yahoo.search.result.Coverage; @@ -15,6 +17,7 @@ import java.util.List; import java.util.Optional; import java.util.OptionalInt; import java.util.Set; +import java.util.concurrent.Callable; /** * @author ollivir @@ -30,6 +33,8 @@ public abstract class InvokerFactory { public abstract Optional<FillInvoker> createFillInvoker(VespaBackEndSearcher searcher, Result result); + public abstract Callable<Pong> createPinger(Node node, ClusterMonitor<Node> monitor); + /** * Create a {@link SearchInvoker} for a list of content nodes. * diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/Client.java b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/Client.java index 4422538cff6..cc37df04a62 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/Client.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/Client.java @@ -2,6 +2,8 @@ package com.yahoo.search.dispatch.rpc; import com.yahoo.compress.CompressionType; +import com.yahoo.compress.Compressor; +import com.yahoo.prelude.Pong; import com.yahoo.prelude.fastsearch.FastHit; import java.util.List; diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcInvokerFactory.java b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcInvokerFactory.java index c1b164aaeaa..b0a418241f8 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcInvokerFactory.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcInvokerFactory.java @@ -1,11 +1,13 @@ // Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.search.dispatch.rpc; +import com.yahoo.prelude.Pong; import com.yahoo.prelude.fastsearch.DocumentDatabase; import com.yahoo.prelude.fastsearch.VespaBackEndSearcher; import com.yahoo.processing.request.CompoundName; import com.yahoo.search.Query; import com.yahoo.search.Result; +import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.dispatch.Dispatcher; import com.yahoo.search.dispatch.FillInvoker; import com.yahoo.search.dispatch.InvokerFactory; @@ -14,6 +16,7 @@ import com.yahoo.search.dispatch.searchcluster.Node; import com.yahoo.search.dispatch.searchcluster.SearchCluster; import java.util.Optional; +import java.util.concurrent.Callable; /** * @author ollivir @@ -23,10 +26,12 @@ public class RpcInvokerFactory extends InvokerFactory { private final static CompoundName dispatchSummaries = new CompoundName("dispatch.summaries"); private final RpcResourcePool rpcResourcePool; + private final boolean dispatchWithProtobuf; - public RpcInvokerFactory(RpcResourcePool rpcResourcePool, SearchCluster searchCluster) { + public RpcInvokerFactory(RpcResourcePool rpcResourcePool, SearchCluster searchCluster, boolean dispatchWithProtobuf) { super(searchCluster); this.rpcResourcePool = rpcResourcePool; + this.dispatchWithProtobuf = dispatchWithProtobuf; } @Override @@ -40,7 +45,7 @@ public class RpcInvokerFactory extends InvokerFactory { boolean summaryNeedsQuery = searcher.summaryNeedsQuery(query); - if(query.properties().getBoolean(Dispatcher.dispatchProtobuf, false)) { + if(query.properties().getBoolean(Dispatcher.dispatchProtobuf, dispatchWithProtobuf)) { return Optional.of(new RpcProtobufFillInvoker(rpcResourcePool, searcher.getDocumentDatabase(query), searcher.getServerId(), summaryNeedsQuery)); } @@ -62,4 +67,9 @@ public class RpcInvokerFactory extends InvokerFactory { public void release() { rpcResourcePool.release(); } + + @Override + public Callable<Pong> createPinger(Node node, ClusterMonitor<Node> monitor) { + return new RpcPing(node, monitor, rpcResourcePool); + } } diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcPing.java b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcPing.java new file mode 100644 index 00000000000..f3479e2e4a9 --- /dev/null +++ b/container-search/src/main/java/com/yahoo/search/dispatch/rpc/RpcPing.java @@ -0,0 +1,77 @@ +package com.yahoo.search.dispatch.rpc; + +import ai.vespa.searchlib.searchprotocol.protobuf.SearchProtocol; +import com.google.protobuf.InvalidProtocolBufferException; +import com.yahoo.compress.CompressionType; +import com.yahoo.compress.Compressor; +import com.yahoo.prelude.Pong; +import com.yahoo.search.cluster.ClusterMonitor; +import com.yahoo.search.dispatch.rpc.Client.ProtobufResponse; +import com.yahoo.search.dispatch.rpc.Client.ResponseOrError; +import com.yahoo.search.dispatch.searchcluster.Node; +import com.yahoo.search.result.ErrorMessage; +import com.yahoo.yolean.Exceptions; + +import java.util.concurrent.Callable; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +public class RpcPing implements Callable<Pong> { + private static final String RPC_METHOD = "vespa.searchprotocol.ping"; + private static final CompressionType PING_COMPRESSION = CompressionType.NONE; + + private final Node node; + private final RpcResourcePool resourcePool; + private final ClusterMonitor<Node> clusterMonitor; + + public RpcPing(Node node, ClusterMonitor<Node> clusterMonitor, RpcResourcePool rpcResourcePool) { + this.node = node; + this.resourcePool = rpcResourcePool; + this.clusterMonitor = clusterMonitor; + } + + @Override + public Pong call() throws Exception { + try { + var queue = new LinkedBlockingQueue<ResponseOrError<ProtobufResponse>>(1); + + sendPing(queue); + + var responseOrError = queue.poll(clusterMonitor.getConfiguration().getRequestTimeout(), TimeUnit.MILLISECONDS); + if (responseOrError == null) { + return new Pong(ErrorMessage.createTimeout("Timed out waiting for pong from " + node)); + } else if (responseOrError.error().isPresent()) { + return new Pong(ErrorMessage.createBackendCommunicationError(responseOrError.error().get())); + } + + return decodeReply(responseOrError.response().get()); + } catch (RuntimeException e) { + return new Pong( + ErrorMessage.createBackendCommunicationError("Exception when pinging " + node + ": " + Exceptions.toMessageString(e))); + } + } + + private void sendPing(LinkedBlockingQueue<ResponseOrError<ProtobufResponse>> queue) { + var connection = resourcePool.nodeConnections().get(node.key()); + var ping = SearchProtocol.MonitorRequest.newBuilder().build().toByteArray(); + double timeoutSeconds = ((double) clusterMonitor.getConfiguration().getRequestTimeout()) / 1000.0; + Compressor.Compression compressionResult = resourcePool.compressor().compress(PING_COMPRESSION, ping); + resourcePool.client().request(RPC_METHOD, connection, compressionResult.type(), ping.length, compressionResult.data(), + rsp -> queue.add(rsp), timeoutSeconds); + } + + private Pong decodeReply(ProtobufResponse response) throws InvalidProtocolBufferException { + CompressionType compression = CompressionType.valueOf(response.compression()); + byte[] responseBytes = resourcePool.compressor().decompress(response.compressedPayload(), compression, response.uncompressedSize()); + var reply = SearchProtocol.MonitorReply.parseFrom(responseBytes); + + if (reply.getDistributionKey() != node.key()) { + return new Pong(ErrorMessage.createBackendCommunicationError( + "Expected pong from node id " + node.key() + ", response is from id " + reply.getDistributionKey())); + } else if (!reply.getOnline()) { + return new Pong(ErrorMessage.createBackendCommunicationError("Node id " + node.key() + " reports being offline")); + } else { + return new Pong(reply.getActiveDocs()); + } + } +} diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Node.java b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Node.java index 7e0e3117628..a71ce0354f9 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Node.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Node.java @@ -53,7 +53,7 @@ public class Node { public boolean isWorking() { return working.get(); } /** Updates the active documents on this node */ - void setActiveDocuments(long activeDocuments) { + public void setActiveDocuments(long activeDocuments) { this.activeDocuments.set(activeDocuments); } diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Pinger.java b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Pinger.java index 7c7a9cb1d1c..dea6f741bb0 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Pinger.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/Pinger.java @@ -15,7 +15,7 @@ import java.util.concurrent.Callable; * @author bratseth * @author ollivir */ -class Pinger implements Callable<Pong> { +public class Pinger implements Callable<Pong> { private final Node node; private final ClusterMonitor<Node> clusterMonitor; private final FS4ResourcePool fs4ResourcePool; @@ -30,8 +30,6 @@ class Pinger implements Callable<Pong> { try { Pong pong = FastSearcher.ping(new Ping(clusterMonitor.getConfiguration().getRequestTimeout()), fs4ResourcePool.getBackend(node.hostname(), node.fs4port()), node.toString()); - if (pong.activeDocuments().isPresent()) - node.setActiveDocuments(pong.activeDocuments().get()); return pong; } catch (RuntimeException e) { return new Pong(ErrorMessage.createBackendCommunicationError("Exception when pinging " + node + ": " diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java index 5c3ef98c523..6c28352f27b 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java @@ -7,12 +7,12 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableMultimap; import com.yahoo.container.handler.VipStatus; import com.yahoo.net.HostName; +import com.yahoo.prelude.Pong; import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.cluster.NodeManager; +import com.yahoo.search.dispatch.InvokerFactory; import com.yahoo.search.result.ErrorMessage; import com.yahoo.vespa.config.search.DispatchConfig; -import com.yahoo.prelude.Pong; -import com.yahoo.prelude.fastsearch.FS4ResourcePool; import java.util.LinkedHashMap; import java.util.List; @@ -46,6 +46,7 @@ public class SearchCluster implements NodeManager<Node> { private final ImmutableList<Group> orderedGroups; private final ClusterMonitor<Node> clusterMonitor; private final VipStatus vipStatus; + private InvokerFactory pingFactory; /** * A search node on this local machine having the entire corpus, which we therefore @@ -57,13 +58,9 @@ public class SearchCluster implements NodeManager<Node> { */ private final Optional<Node> directDispatchTarget; - // Only needed until query requests are moved to rpc - private final FS4ResourcePool fs4ResourcePool; - - public SearchCluster(String clusterId, DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, int containerClusterSize, VipStatus vipStatus) { + public SearchCluster(String clusterId, DispatchConfig dispatchConfig, int containerClusterSize, VipStatus vipStatus) { this.clusterId = clusterId; this.dispatchConfig = dispatchConfig; - this.fs4ResourcePool = fs4ResourcePool; this.vipStatus = vipStatus; List<Node> nodes = toNodes(dispatchConfig); @@ -89,15 +86,20 @@ public class SearchCluster implements NodeManager<Node> { this.directDispatchTarget = findDirectDispatchTarget(HostName.getLocalhost(), size, containerClusterSize, nodesByHost, groups); - // Set up monitoring of the fs4 interface of the nodes - // We can switch to monitoring the rpc interface instead when we move the query phase to rpc this.clusterMonitor = new ClusterMonitor<>(this); - for (Node node : nodes) { - // cluster monitor will only call working() when the - // node transitions from down to up, so we need to - // register the initial (working) state here: - working(node); - clusterMonitor.add(node, true); + } + + public void startClusterMonitoring(InvokerFactory pingFactory) { + this.pingFactory = pingFactory; + + for (var group : orderedGroups) { + for (var node : group.nodes()) { + // cluster monitor will only call working() when the + // node transitions from down to up, so we need to + // register the initial (working) state here: + working(node); + clusterMonitor.add(node, true); + } } } @@ -251,16 +253,21 @@ public class SearchCluster implements NodeManager<Node> { /** Used by the cluster monitor to manage node status */ @Override public void ping(Node node, Executor executor) { - Pinger pinger = new Pinger(node, clusterMonitor, fs4ResourcePool); - FutureTask<Pong> futurePong = new FutureTask<>(pinger); + if (pingFactory == null) // not initialized yet + return; + FutureTask<Pong> futurePong = new FutureTask<>(pingFactory.createPinger(node, clusterMonitor)); executor.execute(futurePong); Pong pong = getPong(futurePong, node); futurePong.cancel(true); - if (pong.badResponse()) + if (pong.badResponse()) { clusterMonitor.failed(node, pong.getError(0)); - else + } else { + if (pong.activeDocuments().isPresent()) { + node.setActiveDocuments(pong.activeDocuments().get()); + } clusterMonitor.responded(node); + } } /** diff --git a/container-search/src/test/java/com/yahoo/fs4/test/QueryTestCase.java b/container-search/src/test/java/com/yahoo/fs4/test/QueryTestCase.java index fc39c1d8fe0..911831e3a65 100644 --- a/container-search/src/test/java/com/yahoo/fs4/test/QueryTestCase.java +++ b/container-search/src/test/java/com/yahoo/fs4/test/QueryTestCase.java @@ -192,6 +192,15 @@ public class QueryTestCase { } @Test + public void testBufferExpands() throws BufferTooSmallException { + Query query = new Query("/?query=chain&sortspec=%2Ba+-b&timeout=0"); + QueryPacket packet = QueryPacket.create("container.0", query); + + ByteBuffer buffer = packet.grantEncodingBuffer(0, ByteBuffer.allocate(2)); + assertEquals(64, buffer.capacity()); + } + + @Test public void testPhraseEqualsPhraseWithPhraseSegment() throws BufferTooSmallException { Query query = new Query(); PhraseItem p = new PhraseItem(); @@ -258,7 +267,6 @@ public class QueryTestCase { assertEqualArrays(correctBuffer,encoded); - packet.allocateAndEncode(0x07070707); buffer = packet.grantEncodingBuffer(0x09090909); correctBuffer = new byte[] {0,0,0,46,0,0,0,-38,9,9,9,9, // Header 0,0,0,6, // Features diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java b/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java index 859e10dbe2c..25aed879a48 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java @@ -137,7 +137,7 @@ public class DispatcherTest { public class MockRpcInvokerFactory extends RpcInvokerFactory { public MockRpcInvokerFactory() { - super(null, null); + super(null, null, true); } @Override diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/LoadBalancerTest.java b/container-search/src/test/java/com/yahoo/search/dispatch/LoadBalancerTest.java index 141d87a41ab..1ebf7940f25 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/LoadBalancerTest.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/LoadBalancerTest.java @@ -29,7 +29,7 @@ public class LoadBalancerTest { @Test public void requireThatLoadBalancerServesSingleNodeSetups() { Node n1 = new Node(0, "test-node1", 0, 0); - SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1), null, 1, null); + SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1), 1, null); LoadBalancer lb = new LoadBalancer(cluster, true); Optional<Group> grp = lb.takeGroup(null); @@ -43,7 +43,7 @@ public class LoadBalancerTest { public void requireThatLoadBalancerServesMultiGroupSetups() { Node n1 = new Node(0, "test-node1", 0, 0); Node n2 = new Node(1, "test-node2", 1, 1); - SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2), null, 1, null); + SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2), 1, null); LoadBalancer lb = new LoadBalancer(cluster, true); Optional<Group> grp = lb.takeGroup(null); @@ -59,7 +59,7 @@ public class LoadBalancerTest { Node n2 = new Node(1, "test-node2", 1, 0); Node n3 = new Node(0, "test-node3", 0, 1); Node n4 = new Node(1, "test-node4", 1, 1); - SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2, n3, n4), null, 2, null); + SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2, n3, n4), 2, null); LoadBalancer lb = new LoadBalancer(cluster, true); Optional<Group> grp = lb.takeGroup(null); @@ -70,7 +70,7 @@ public class LoadBalancerTest { public void requireThatLoadBalancerReturnsDifferentGroups() { Node n1 = new Node(0, "test-node1", 0, 0); Node n2 = new Node(1, "test-node2", 1, 1); - SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2), null, 1, null); + SearchCluster cluster = new SearchCluster("a", createDispatchConfig(n1, n2), 1, null); LoadBalancer lb = new LoadBalancer(cluster, true); // get first group diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/MockSearchCluster.java b/container-search/src/test/java/com/yahoo/search/dispatch/MockSearchCluster.java index 3919bc26bdc..42a22f6f86b 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/MockSearchCluster.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/MockSearchCluster.java @@ -28,7 +28,7 @@ public class MockSearchCluster extends SearchCluster { } public MockSearchCluster(String clusterId, DispatchConfig dispatchConfig, int groups, int nodesPerGroup) { - super(clusterId, dispatchConfig, null, 1, null); + super(clusterId, dispatchConfig, 1, null); ImmutableList.Builder<Group> orderedGroupBuilder = ImmutableList.builder(); ImmutableMap.Builder<Integer, Group> groupBuilder = ImmutableMap.builder(); diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/rpc/FillTestCase.java b/container-search/src/test/java/com/yahoo/search/dispatch/rpc/FillTestCase.java index 2adbd12a2aa..e059008acac 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/rpc/FillTestCase.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/rpc/FillTestCase.java @@ -39,7 +39,7 @@ public class FillTestCase { nodes.put(1, client.createConnection("host1", 123)); nodes.put(2, client.createConnection("host2", 123)); RpcResourcePool rpcResourcePool = new RpcResourcePool(client, nodes); - RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null); + RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null, true); Query query = new Query(); Result result = new Result(query); @@ -76,7 +76,7 @@ public class FillTestCase { nodes.put(1, client.createConnection("host1", 123)); nodes.put(2, client.createConnection("host2", 123)); RpcResourcePool rpcResourcePool = new RpcResourcePool(client, nodes); - RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null); + RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null, true); Query query = new Query(); Result result = new Result(query); @@ -116,7 +116,7 @@ public class FillTestCase { Map<Integer, Client.NodeConnection> nodes = new HashMap<>(); nodes.put(0, client.createConnection("host0", 123)); RpcResourcePool rpcResourcePool = new RpcResourcePool(client, nodes); - RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null); + RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null, true); Query query = new Query(); Result result = new Result(query); @@ -134,7 +134,7 @@ public class FillTestCase { Map<Integer, Client.NodeConnection> nodes = new HashMap<>(); nodes.put(0, client.createConnection("host0", 123)); RpcResourcePool rpcResourcePool = new RpcResourcePool(client, nodes); - RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null); + RpcInvokerFactory factory = new RpcInvokerFactory(rpcResourcePool, null, true); Query query = new Query(); Result result = new Result(query); diff --git a/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceAllocation.java b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceAllocation.java new file mode 100644 index 00000000000..3b86b7b8219 --- /dev/null +++ b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceAllocation.java @@ -0,0 +1,50 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.api.integration.resource; + +import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryNode; + +import java.util.List; + +/** + * Stores the total amount of resources allocated to a list of nodes + * + * @author leandroalves + */ +public class ResourceAllocation { + + private final double cpuCores; + private final double memoryGb; + private final double diskGb; + + private ResourceAllocation(double cpuCores, double memoryGb, double diskGb) { + this.cpuCores = cpuCores; + this.memoryGb = memoryGb; + this.diskGb = diskGb; + } + + public static ResourceAllocation from(List<NodeRepositoryNode> nodes) { + return new ResourceAllocation( + nodes.stream().mapToDouble(NodeRepositoryNode::getMinCpuCores).sum(), + nodes.stream().mapToDouble(NodeRepositoryNode::getMinMainMemoryAvailableGb).sum(), + nodes.stream().mapToDouble(NodeRepositoryNode::getMinDiskAvailableGb).sum() + ); + } + + public double usageFraction(ResourceAllocation total) { + return (cpuCores / total.cpuCores + memoryGb / total.memoryGb + diskGb / total.diskGb) / 3; + } + + public double getCpuCores() { + return cpuCores; + } + + public double getMemoryGb() { + return memoryGb; + } + + public double getDiskGb() { + return diskGb; + } + +} + diff --git a/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshot.java b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshot.java new file mode 100644 index 00000000000..7f7a6b758d5 --- /dev/null +++ b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshot.java @@ -0,0 +1,27 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.api.integration.resource; + +import java.time.Instant; + +/** + * @author olaa + */ +public class ResourceSnapshot { + + private final ResourceAllocation resourceAllocation; + private final Instant timestamp; + + public ResourceSnapshot(ResourceAllocation resourceAllocation, Instant timestamp) { + this.resourceAllocation = resourceAllocation; + this.timestamp = timestamp; + } + + public ResourceAllocation getResourceAllocation() { + return resourceAllocation; + } + + public Instant getTimestamp() { + return timestamp; + } + +} diff --git a/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshotConsumer.java b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshotConsumer.java new file mode 100644 index 00000000000..f7f3eddb482 --- /dev/null +++ b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/ResourceSnapshotConsumer.java @@ -0,0 +1,16 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.api.integration.resource; + +import com.yahoo.config.provision.ApplicationId; + +import java.util.Map; + +/** + * Consumes a snapshot of resourses allocated/used per application. + * + * @author olaa + */ +public interface ResourceSnapshotConsumer { + + public void consume(Map<ApplicationId, ResourceSnapshot> resources); +} diff --git a/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/package-info.java b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/package-info.java new file mode 100644 index 00000000000..bfb44eb94ba --- /dev/null +++ b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/resource/package-info.java @@ -0,0 +1,5 @@ +// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +@ExportPackage +package com.yahoo.vespa.hosted.controller.api.integration.resource; + +import com.yahoo.osgi.annotation.ExportPackage; diff --git a/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/stubs/MockResourceSnapshotConsumer.java b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/stubs/MockResourceSnapshotConsumer.java new file mode 100644 index 00000000000..d5d7b63e933 --- /dev/null +++ b/controller-api/src/main/java/com/yahoo/vespa/hosted/controller/api/integration/stubs/MockResourceSnapshotConsumer.java @@ -0,0 +1,25 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.api.integration.stubs; + +import com.yahoo.config.provision.ApplicationId; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshot; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshotConsumer; + +import java.util.Map; + +/** + * @author olaa + */ +public class MockResourceSnapshotConsumer implements ResourceSnapshotConsumer { + + private Map<ApplicationId, ResourceSnapshot> resources; + + @Override + public void consume(Map<ApplicationId, ResourceSnapshot> resources){ + this.resources = resources; + } + + public Map<ApplicationId, ResourceSnapshot> consumedResources() { + return resources; + } +} diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ControllerMaintenance.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ControllerMaintenance.java index cc9e4020dab..506231f086b 100644 --- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ControllerMaintenance.java +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ControllerMaintenance.java @@ -5,6 +5,7 @@ import com.yahoo.component.AbstractComponent; import com.yahoo.jdisc.Metric; import com.yahoo.vespa.hosted.controller.Controller; import com.yahoo.vespa.hosted.controller.api.integration.organization.ContactRetriever; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshotConsumer; import com.yahoo.vespa.hosted.controller.authority.config.ApiAuthorityConfig; import com.yahoo.vespa.hosted.controller.api.integration.chef.Chef; import com.yahoo.vespa.hosted.controller.api.integration.dns.NameService; @@ -53,6 +54,7 @@ public class ControllerMaintenance extends AbstractComponent { private final ContactInformationMaintainer contactInformationMaintainer; private final CostReportMaintainer costReportMaintainer; private final RoutingPolicyMaintainer routingPolicyMaintainer; + private final ResourceMeterMaintainer resourceMeterMaintainer; @SuppressWarnings("unused") // instantiated by Dependency Injection public ControllerMaintenance(MaintainerConfig maintainerConfig, ApiAuthorityConfig apiAuthorityConfig, Controller controller, CuratorDb curator, @@ -61,6 +63,7 @@ public class ControllerMaintenance extends AbstractComponent { NameService nameService, NodeRepositoryClientInterface nodeRepositoryClient, ContactRetriever contactRetriever, CostReportConsumer reportConsumer, + ResourceSnapshotConsumer resourceSnapshotConsumer, SelfHostedCostConfig selfHostedCostConfig) { Duration maintenanceInterval = Duration.ofMinutes(maintainerConfig.intervalMinutes()); this.jobControl = jobControl; @@ -83,6 +86,7 @@ public class ControllerMaintenance extends AbstractComponent { contactInformationMaintainer = new ContactInformationMaintainer(controller, Duration.ofHours(12), jobControl, contactRetriever); costReportMaintainer = new CostReportMaintainer(controller, Duration.ofHours(2), reportConsumer, jobControl, nodeRepositoryClient, Clock.systemUTC(), selfHostedCostConfig); routingPolicyMaintainer = new RoutingPolicyMaintainer(controller, Duration.ofMinutes(5), jobControl, nameService, curator); + resourceMeterMaintainer = new ResourceMeterMaintainer(controller, Duration.ofMinutes(5), jobControl, nodeRepositoryClient, Clock.systemUTC(), metric, resourceSnapshotConsumer); } public Upgrader upgrader() { return upgrader; } @@ -111,6 +115,7 @@ public class ControllerMaintenance extends AbstractComponent { contactInformationMaintainer.deconstruct(); costReportMaintainer.deconstruct(); routingPolicyMaintainer.deconstruct(); + resourceMeterMaintainer.deconstruct(); } /** Create one OS upgrader per cloud found in the zone registry of controller */ diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainer.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainer.java index 2298c3c92fe..2b26e93aeb8 100644 --- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainer.java +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainer.java @@ -2,6 +2,7 @@ package com.yahoo.vespa.hosted.controller.maintenance; import com.google.inject.Inject; +import com.yahoo.config.provision.CloudName; import com.yahoo.config.provision.SystemName; import com.yahoo.vespa.hosted.controller.Controller; import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryClientInterface; @@ -46,6 +47,6 @@ public class CostReportMaintainer extends Maintainer { @Override protected void maintain() { - consumer.Consume(CostCalculator.toCsv(CostCalculator.calculateCost(nodeRepository, controller(), clock, selfHostedCostConfig))); + consumer.Consume(CostCalculator.resourceShareByPropertyToCsv(nodeRepository, controller(), clock, selfHostedCostConfig, CloudName.from("yahoo"))); } } diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainer.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainer.java new file mode 100644 index 00000000000..5ed7a14836e --- /dev/null +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainer.java @@ -0,0 +1,104 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.maintenance; + +import com.yahoo.config.provision.SystemName; +import com.yahoo.jdisc.Metric; +import com.yahoo.vespa.hosted.controller.Controller; +import com.yahoo.config.provision.ApplicationId; +import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeOwner; +import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryClientInterface; +import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryNode; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshot; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshotConsumer; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceAllocation; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.*; +import java.util.stream.Collectors; + +import static com.yahoo.yolean.Exceptions.uncheck; + +/** + * Creates a ResourceSnapshot per application, which is then passed on to a ResourceSnapshotConsumer + * TODO: Write JSON blob of node repo somewhere + * + * @author olaa + */ +public class ResourceMeterMaintainer extends Maintainer { + + private final Clock clock; + private final Metric metric; + private final NodeRepositoryClientInterface nodeRepository; + private final ResourceSnapshotConsumer resourceSnapshotConsumer; + + private static final String metering_last_reported = "metering_last_reported"; + private static final String metering_total_reported = "metering_total_reported"; + + @SuppressWarnings("WeakerAccess") + public ResourceMeterMaintainer(Controller controller, + Duration interval, + JobControl jobControl, + NodeRepositoryClientInterface nodeRepository, + Clock clock, + Metric metric, + ResourceSnapshotConsumer resourceSnapshotConsumer) { + super(controller, interval, jobControl, ResourceMeterMaintainer.class.getSimpleName(), Set.of(SystemName.cd)); + this.clock = clock; + this.nodeRepository = nodeRepository; + this.metric = metric; + this.resourceSnapshotConsumer = resourceSnapshotConsumer; + } + + @Override + protected void maintain() { + List<NodeRepositoryNode> nodes = getNodes(); + Map<ApplicationId, ResourceAllocation> resourceAllocationByApplication = getResourceAllocationByApplication(nodes); + + // For now, we're only interested in resource allocation + Instant timeStamp = clock.instant(); + Map<ApplicationId, ResourceSnapshot> resourceSnapshots = resourceAllocationByApplication.entrySet().stream() + .collect(Collectors.toMap( + e -> e.getKey(), + e -> new ResourceSnapshot(e.getValue(), timeStamp)) + ); + + + resourceSnapshotConsumer.consume(resourceSnapshots); + + metric.set(metering_last_reported, clock.millis() / 1000, metric.createContext(Collections.emptyMap())); + metric.set(metering_total_reported, resourceSnapshots.values().stream() + .map(ResourceSnapshot::getResourceAllocation) + .mapToDouble(r -> r.getCpuCores() + r.getMemoryGb() + r.getDiskGb()) // total metered resource usage, for alerting on drastic changes + .sum() + , metric.createContext(Collections.emptyMap())); + } + + private List<NodeRepositoryNode> getNodes() { + return controller().zoneRegistry().zones() + .reachable().ids().stream() + .flatMap(zoneId -> uncheck(() -> nodeRepository.listNodes(zoneId, true).nodes().stream())) + .filter(node -> node.getOwner() != null && !node.getOwner().getTenant().equals("hosted-vespa")) + .collect(Collectors.toList()); + } + + private Map<ApplicationId, ResourceAllocation> getResourceAllocationByApplication(List<NodeRepositoryNode> nodes) { + Map<ApplicationId, List<NodeRepositoryNode>> applicationNodes = new HashMap<>(); + + nodes.stream().forEach(node -> applicationNodes.computeIfAbsent(applicationIdFromNodeOwner(node.getOwner()), n -> new ArrayList<>()).add(node)); + + return applicationNodes.entrySet().stream() + .collect( + Collectors.toMap( + entry -> entry.getKey(), + entry -> ResourceAllocation.from(entry.getValue()) + ) + ); + } + + private ApplicationId applicationIdFromNodeOwner(NodeOwner owner) { + return ApplicationId.from(owner.getTenant(), owner.getApplication(), owner.getInstance()); + } + +} diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/ApplicationApiHandler.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/ApplicationApiHandler.java index 69a4216d221..402f91f1a14 100644 --- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/ApplicationApiHandler.java +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/application/ApplicationApiHandler.java @@ -110,6 +110,8 @@ import static java.util.stream.Collectors.joining; @SuppressWarnings("unused") // created by injection public class ApplicationApiHandler extends LoggingRequestHandler { + private static final String OPTIONAL_PREFIX = "/api"; + private final Controller controller; private final AccessControlRequests accessControlRequests; @@ -162,7 +164,7 @@ public class ApplicationApiHandler extends LoggingRequestHandler { } private HttpResponse handleGET(HttpRequest request) { - Path path = new Path(request.getUri().getPath()); + Path path = new Path(request.getUri().getPath(), OPTIONAL_PREFIX); if (path.matches("/application/v4/")) return root(request); if (path.matches("/application/v4/user")) return authenticatedUser(request); if (path.matches("/application/v4/tenant")) return tenants(request); @@ -185,7 +187,7 @@ public class ApplicationApiHandler extends LoggingRequestHandler { } private HttpResponse handlePUT(HttpRequest request) { - Path path = new Path(request.getUri().getPath()); + Path path = new Path(request.getUri().getPath(), OPTIONAL_PREFIX); if (path.matches("/application/v4/user")) return createUser(request); if (path.matches("/application/v4/tenant/{tenant}")) return updateTenant(path.get("tenant"), request); if (path.matches("/application/v4/tenant/{tenant}/application/{application}/environment/{environment}/region/{region}/instance/{instance}/global-rotation/override")) @@ -194,7 +196,7 @@ public class ApplicationApiHandler extends LoggingRequestHandler { } private HttpResponse handlePOST(HttpRequest request) { - Path path = new Path(request.getUri().getPath()); + Path path = new Path(request.getUri().getPath(), OPTIONAL_PREFIX); if (path.matches("/application/v4/tenant/{tenant}")) return createTenant(path.get("tenant"), request); if (path.matches("/application/v4/tenant/{tenant}/application/{application}")) return createApplication(path.get("tenant"), path.get("application"), request); if (path.matches("/application/v4/tenant/{tenant}/application/{application}/promote")) return promoteApplication(path.get("tenant"), path.get("application"), request); @@ -213,14 +215,14 @@ public class ApplicationApiHandler extends LoggingRequestHandler { } private HttpResponse handlePATCH(HttpRequest request) { - Path path = new Path(request.getUri().getPath()); + Path path = new Path(request.getUri().getPath(), OPTIONAL_PREFIX); if (path.matches("/application/v4/tenant/{tenant}/application/{application}")) return setMajorVersion(path.get("tenant"), path.get("application"), request); return ErrorResponse.notFoundError("Nothing at " + path); } private HttpResponse handleDELETE(HttpRequest request) { - Path path = new Path(request.getUri().getPath()); + Path path = new Path(request.getUri().getPath(), OPTIONAL_PREFIX); if (path.matches("/application/v4/tenant/{tenant}")) return deleteTenant(path.get("tenant"), request); if (path.matches("/application/v4/tenant/{tenant}/application/{application}")) return deleteApplication(path.get("tenant"), path.get("application"), request); if (path.matches("/application/v4/tenant/{tenant}/application/{application}/deploying")) return cancelDeploy(path.get("tenant"), path.get("application"), "all"); diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostApiHandler.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostApiHandler.java index 6d599d32cc6..a82d2f22e74 100644 --- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostApiHandler.java +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostApiHandler.java @@ -1,5 +1,6 @@ package com.yahoo.vespa.hosted.controller.restapi.cost; +import com.yahoo.config.provision.CloudName; import com.yahoo.container.jdisc.HttpRequest; import com.yahoo.container.jdisc.HttpResponse; import com.yahoo.container.jdisc.LoggingRequestHandler; @@ -11,6 +12,7 @@ import com.yahoo.vespa.hosted.controller.restapi.StringResponse; import com.yahoo.vespa.hosted.controller.restapi.cost.config.SelfHostedCostConfig; import java.time.Clock; +import java.util.Optional; import static com.yahoo.jdisc.http.HttpRequest.Method.GET; @@ -36,7 +38,9 @@ public class CostApiHandler extends LoggingRequestHandler { Path path = new Path(request.getUri().getPath()); if (path.matches("/cost/v1/csv")) { - return new StringResponse(CostCalculator.toCsv(CostCalculator.calculateCost(nodeRepository, controller, Clock.systemUTC(), selfHostedCostConfig))); + Optional<String> cloudProperty = Optional.ofNullable(request.getProperty("cloud")); + CloudName cloud = cloudProperty.map(CloudName::from).orElse(CloudName.defaultName()); + return new StringResponse(CostCalculator.resourceShareByPropertyToCsv(nodeRepository, controller, Clock.systemUTC(), selfHostedCostConfig, cloud)); } return ErrorResponse.notFoundError("Nothing at " + path); diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostCalculator.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostCalculator.java index 88fe28a3613..18c00d69b62 100644 --- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostCalculator.java +++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/cost/CostCalculator.java @@ -7,6 +7,7 @@ import com.yahoo.vespa.hosted.controller.api.identifiers.Property; import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeOwner; import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryClientInterface; import com.yahoo.vespa.hosted.controller.api.integration.noderepository.NodeRepositoryNode; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceAllocation; import com.yahoo.vespa.hosted.controller.restapi.cost.config.SelfHostedCostConfig; import com.yahoo.vespa.hosted.controller.tenant.AthenzTenant; @@ -24,33 +25,35 @@ public class CostCalculator { private static final double SELF_HOSTED_DISCOUNT = .5; - public static Map<Property, ResourceAllocation> calculateCost(NodeRepositoryClientInterface nodeRepository, - Controller controller, - Clock clock, - SelfHostedCostConfig selfHostedCostConfig) { + public static String resourceShareByPropertyToCsv(NodeRepositoryClientInterface nodeRepository, + Controller controller, + Clock clock, + SelfHostedCostConfig selfHostedCostConfig, + CloudName cloudName) { String date = LocalDate.now(clock).toString(); List<NodeRepositoryNode> nodes = controller.zoneRegistry().zones() - .reachable().in(Environment.prod).ofCloud(CloudName.from("yahoo")).ids().stream() + .reachable().in(Environment.prod).ofCloud(cloudName).ids().stream() .flatMap(zoneId -> uncheck(() -> nodeRepository.listNodes(zoneId, true).nodes().stream())) .filter(node -> node.getOwner() != null && !node.getOwner().getTenant().equals("hosted-vespa")) .collect(Collectors.toList()); - selfHostedCostConfig.properties().stream().map(property -> { - NodeRepositoryNode selfHostedNode = new NodeRepositoryNode(); + if (cloudName.equals(CloudName.from("yahoo"))) + selfHostedCostConfig.properties().stream().map(property -> { + NodeRepositoryNode selfHostedNode = new NodeRepositoryNode(); - NodeOwner owner = new NodeOwner(); - owner.tenant = property.name(); - selfHostedNode.setOwner(owner); - selfHostedNode.setMinCpuCores(property.cpuCores() * SELF_HOSTED_DISCOUNT); - selfHostedNode.setMinMainMemoryAvailableGb(property.memoryGb() * SELF_HOSTED_DISCOUNT); - selfHostedNode.setMinDiskAvailableGb(property.diskGb() * SELF_HOSTED_DISCOUNT); + NodeOwner owner = new NodeOwner(); + owner.tenant = property.name(); + selfHostedNode.setOwner(owner); + selfHostedNode.setMinCpuCores(property.cpuCores() * SELF_HOSTED_DISCOUNT); + selfHostedNode.setMinMainMemoryAvailableGb(property.memoryGb() * SELF_HOSTED_DISCOUNT); + selfHostedNode.setMinDiskAvailableGb(property.diskGb() * SELF_HOSTED_DISCOUNT); - return selfHostedNode; - }).forEach(nodes::add); + return selfHostedNode; + }).forEach(nodes::add); - ResourceAllocation total = ResourceAllocation.from(date, nodes, null); + ResourceAllocation totalResourceAllocation = ResourceAllocation.from(nodes); Map<String, Property> propertyByTenantName = controller.tenants().asList().stream() .filter(AthenzTenant.class::isInstance) @@ -63,58 +66,29 @@ public class CostCalculator { .map(SelfHostedCostConfig.Properties::name) .forEach(name -> propertyByTenantName.put(name, new Property(name))); - return nodes.stream() + Map<Property, ResourceAllocation> resourceShareByProperty = nodes.stream() .filter(node -> propertyByTenantName.containsKey(node.getOwner().tenant)) .collect(Collectors.groupingBy( node -> propertyByTenantName.get(node.getOwner().tenant), Collectors.collectingAndThen( Collectors.toList(), - (tenantNodes) -> ResourceAllocation.from(date, tenantNodes, total) + ResourceAllocation::from ) )); - } - static class ResourceAllocation { - final double cpuCores; - final double memoryGb; - final double diskGb; - final String date; - final ResourceAllocation total; - - private ResourceAllocation(String date, double cpuCores, double memoryGb, double diskGb, ResourceAllocation total) { - this.date = date; - this.cpuCores = cpuCores; - this.memoryGb = memoryGb; - this.diskGb = diskGb; - this.total = total; - } - - private static ResourceAllocation from(String date, List<NodeRepositoryNode> nodes, ResourceAllocation total) { - return new ResourceAllocation( - date, - nodes.stream().mapToDouble(NodeRepositoryNode::getMinCpuCores).sum(), - nodes.stream().mapToDouble(NodeRepositoryNode::getMinMainMemoryAvailableGb).sum(), - nodes.stream().mapToDouble(NodeRepositoryNode::getMinDiskAvailableGb).sum(), - total - ); - } - - private double usageFraction() { - return (cpuCores / total.cpuCores + memoryGb / total.memoryGb + diskGb / total.diskGb) / 3; - } + return toCsv(resourceShareByProperty, date, totalResourceAllocation); } - public static String toCsv(Map<Property, ResourceAllocation> resourceShareByProperty) { + private static String toCsv(Map<Property, ResourceAllocation> resourceShareByProperty, String date, ResourceAllocation totalResourceAllocation) { String header = "Date,Property,Reserved Cpu Cores,Reserved Memory GB,Reserved Disk Space GB,Usage Fraction\n"; String entries = resourceShareByProperty.entrySet().stream() - .sorted((Comparator.comparingDouble(entry -> entry.getValue().usageFraction()))) + .sorted((Comparator.comparingDouble(entry -> entry.getValue().usageFraction(totalResourceAllocation)))) .map(propertyEntry -> { ResourceAllocation r = propertyEntry.getValue(); - return Stream.of(r.date, propertyEntry.getKey(), r.cpuCores, r.memoryGb, r.diskGb, r.usageFraction()) + return Stream.of(date, propertyEntry.getKey(), r.getCpuCores(), r.getMemoryGb(), r.getDiskGb(), r.usageFraction(totalResourceAllocation)) .map(Object::toString).collect(Collectors.joining(",")); }) .collect(Collectors.joining("\n")); return header + entries; } - } diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/integration/NodeRepositoryClientMock.java b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/integration/NodeRepositoryClientMock.java index daddc46589d..ccd09cb9261 100644 --- a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/integration/NodeRepositoryClientMock.java +++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/integration/NodeRepositoryClientMock.java @@ -56,8 +56,8 @@ public class NodeRepositoryClientMock implements NodeRepositoryClientInterface { node.setMinDiskAvailableGb(500d); node.setMinMainMemoryAvailableGb(24d); NodeOwner owner = new NodeOwner(); - owner.tenant = "lsbe"; - owner.application = "local-search"; + owner.tenant = "tenant1"; + owner.application = "app1"; owner.instance = "default"; node.setOwner(owner); NodeMembership membership = new NodeMembership(); @@ -76,8 +76,8 @@ public class NodeRepositoryClientMock implements NodeRepositoryClientInterface { node.setMinDiskAvailableGb(500d); node.setMinMainMemoryAvailableGb(24d); NodeOwner owner = new NodeOwner(); - owner.tenant = "mediasearch"; - owner.application = "imagesearch"; + owner.tenant = "tenant2"; + owner.application = "app2"; owner.instance = "default"; node.setOwner(owner); NodeMembership membership = new NodeMembership(); diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainerTest.java b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainerTest.java index 01f3f55c679..890af974a0e 100644 --- a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainerTest.java +++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/CostReportMaintainerTest.java @@ -34,8 +34,8 @@ public class CostReportMaintainerTest { .build(); - tester.createTenant("lsbe", "local-search", 1L); - tester.createTenant("mediasearch", "msbe", 2L); + tester.createTenant("tenant1", "app1", 1L); + tester.createTenant("tenant2", "app2", 2L); CostReportMaintainer maintainer = new CostReportMaintainer( tester.controller(), Duration.ofDays(1), diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainerTest.java b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainerTest.java new file mode 100644 index 00000000000..df2a4b5ca7f --- /dev/null +++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/maintenance/ResourceMeterMaintainerTest.java @@ -0,0 +1,50 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.controller.maintenance; + +import com.yahoo.vespa.hosted.controller.ControllerTester; +import com.yahoo.config.provision.ApplicationId; +import com.yahoo.vespa.hosted.controller.api.integration.resource.ResourceSnapshot; +import com.yahoo.vespa.hosted.controller.api.integration.stubs.MockResourceSnapshotConsumer; +import com.yahoo.vespa.hosted.controller.integration.NodeRepositoryClientMock; +import com.yahoo.vespa.hosted.controller.integration.MetricsMock; +import org.junit.Test; + +import java.time.Duration; +import java.util.Map; + +import static org.junit.Assert.*; + +/** + * @author olaa + */ +public class ResourceMeterMaintainerTest { + + private final double DELTA = Double.MIN_VALUE; + private NodeRepositoryClientMock nodeRepository = new NodeRepositoryClientMock(); + private MockResourceSnapshotConsumer snapshotConsumer = new MockResourceSnapshotConsumer(); + private MetricsMock metrics = new MetricsMock(); + + @Test + public void testMaintainer() { + ControllerTester tester = new ControllerTester(); + ResourceMeterMaintainer resourceMeterMaintainer = new ResourceMeterMaintainer(tester.controller(), Duration.ofMinutes(5), new JobControl(tester.curator()), nodeRepository, tester.clock(), metrics, snapshotConsumer); + resourceMeterMaintainer.maintain(); + Map<ApplicationId, ResourceSnapshot> consumedResources = snapshotConsumer.consumedResources(); + + assertEquals(2, consumedResources.size()); + + ResourceSnapshot app1 = consumedResources.get(ApplicationId.from("tenant1", "app1", "default")); + ResourceSnapshot app2 = consumedResources.get(ApplicationId.from("tenant2", "app2", "default")); + + assertEquals(96, app1.getResourceAllocation().getCpuCores(), DELTA); + assertEquals(96, app1.getResourceAllocation().getMemoryGb(), DELTA); + assertEquals(2000, app1.getResourceAllocation().getDiskGb(), DELTA); + + assertEquals(160, app2.getResourceAllocation().getCpuCores(), DELTA); + assertEquals(96, app2.getResourceAllocation().getMemoryGb(), DELTA); + assertEquals(2000, app2.getResourceAllocation().getDiskGb(), DELTA); + + assertEquals(tester.clock().millis()/1000, metrics.getMetric("metering_last_reported")); + assertEquals(4448.0d, (Double) metrics.getMetric("metering_total_reported"), DELTA); + } +}
\ No newline at end of file diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/ControllerContainerTest.java b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/ControllerContainerTest.java index 71b65770b1e..331a6ba9ac8 100644 --- a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/ControllerContainerTest.java +++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/ControllerContainerTest.java @@ -75,6 +75,7 @@ public class ControllerContainerTest { " <component id='com.yahoo.vespa.hosted.controller.api.integration.stubs.MockRunDataStore'/>\n" + " <component id='com.yahoo.vespa.hosted.controller.api.integration.organization.MockContactRetriever'/>\n" + " <component id='com.yahoo.vespa.hosted.controller.api.integration.organization.MockIssueHandler'/>\n" + + " <component id='com.yahoo.vespa.hosted.controller.api.integration.stubs.MockResourceSnapshotConsumer'/>\n" + " <component id='com.yahoo.vespa.hosted.controller.integration.ConfigServerMock'/>\n" + " <component id='com.yahoo.vespa.hosted.controller.integration.NodeRepositoryClientMock'/>\n" + " <component id='com.yahoo.vespa.hosted.controller.integration.ZoneRegistryMock'/>\n" + diff --git a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/controller/responses/maintenance.json b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/controller/responses/maintenance.json index dd64d480453..0b5d3912214 100644 --- a/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/controller/responses/maintenance.json +++ b/controller-server/src/test/java/com/yahoo/vespa/hosted/controller/restapi/controller/responses/maintenance.json @@ -46,6 +46,9 @@ "name": "ReadyJobsTrigger" }, { + "name": "ResourceMeterMaintainer" + }, + { "name": "RoutingPolicyMaintainer" }, { diff --git a/document/src/tests/CMakeLists.txt b/document/src/tests/CMakeLists.txt index b78fd66b687..bb668287a8c 100644 --- a/document/src/tests/CMakeLists.txt +++ b/document/src/tests/CMakeLists.txt @@ -54,9 +54,7 @@ vespa_add_executable(document_testrunner_app TEST document_documentconfig ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME document_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:document_testrunner_app> - DEPENDS document_testrunner_app + COMMAND document_testrunner_app ) diff --git a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java index 9de9f8f2ba4..be564098121 100644 --- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java +++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java @@ -95,6 +95,12 @@ public class Flags { "Takes effect at redeployment", APPLICATION_ID); + public static final UnboundBooleanFlag DISPATCH_WITH_PROTOBUF = defineFeatureFlag( + "dispatch-with-protobuf", false, + "Should the java dispatcher use protobuf/jrt as the default", + "Takes effect at redeployment", + APPLICATION_ID); + public static final UnboundBooleanFlag ENABLE_DYNAMIC_PROVISIONING = defineFeatureFlag( "enable-dynamic-provisioning", false, "Provision a new docker host when we otherwise can't allocate a docker node", @@ -107,12 +113,6 @@ public class Flags { "Takes effect on next node agent tick (but does not clear existing failure reports)", HOSTNAME); - public static final UnboundBooleanFlag USE_SEPARATE_SERVICE_TYPE_FOR_LOGSERVER_CONTAINER = defineFeatureFlag( - "use-separate-service-type-for-logserver-container", true, - "Use separate service type for Logserver container, resulting in logserver container not being an application endpoint", - "Takes effect at redeployment", - APPLICATION_ID); - public static final UnboundBooleanFlag ENABLE_METRICS_PROXY_CONTAINER = defineFeatureFlag( "enable-metrics-proxy-container", false, "Start a container for metrics-proxy on every vespa node", diff --git a/logd/CMakeLists.txt b/logd/CMakeLists.txt index 9a5fdf32841..e16d3bd5179 100644 --- a/logd/CMakeLists.txt +++ b/logd/CMakeLists.txt @@ -17,6 +17,7 @@ vespa_define_module( src/tests/legacy_forwarder src/tests/proto_converter src/tests/rotate + src/tests/rpc_forwarder ) vespa_install_script(src/apps/retention/retention-enforcer.sh vespa-retention-enforcer sbin) diff --git a/logd/src/logd/CMakeLists.txt b/logd/src/logd/CMakeLists.txt index baf52f1d5d8..629c7b1637a 100644 --- a/logd/src/logd/CMakeLists.txt +++ b/logd/src/logd/CMakeLists.txt @@ -16,6 +16,7 @@ vespa_add_library(logd STATIC legacy_forwarder.cpp metrics.cpp proto_converter.cpp + rpc_forwarder.cpp state_reporter.cpp watcher.cpp ${logd_PROTOBUF_SRCS} diff --git a/logd/src/logd/exceptions.h b/logd/src/logd/exceptions.h index b2ff3516b69..82e8b570c3b 100644 --- a/logd/src/logd/exceptions.h +++ b/logd/src/logd/exceptions.h @@ -18,7 +18,12 @@ public: class ConnectionException : public MsgException { public: - ConnectionException(const char *s) : MsgException(s) {} + ConnectionException(const std::string& msg) : MsgException(msg) {} +}; + +class DecodeException : public MsgException { +public: + DecodeException(const std::string& msg) : MsgException(msg) {} }; class SigTermException : public MsgException diff --git a/logd/src/logd/forwarder.h b/logd/src/logd/forwarder.h index a0a1c5f1ea5..93cfdb3de9f 100644 --- a/logd/src/logd/forwarder.h +++ b/logd/src/logd/forwarder.h @@ -2,8 +2,15 @@ #pragma once +#include <vespa/log/log.h> +#include <map> +#include <string_view> + namespace logdemon { +// Mapping saying if a level should be forwarded or not +using ForwardMap = std::map<ns_log::Logger::LogLevel, bool>; + /** * Interface used to forward log lines to something. */ @@ -11,7 +18,8 @@ class Forwarder { public: virtual ~Forwarder() {} virtual void sendMode() = 0; - virtual void forwardLine(const char *line, const char *eol) = 0; + virtual void forwardLine(std::string_view log_line) = 0; + virtual void flush() = 0; virtual int badLines() const = 0; virtual void resetBadLines() = 0; }; diff --git a/logd/src/logd/legacy_forwarder.cpp b/logd/src/logd/legacy_forwarder.cpp index b8b93a03530..6b2f430d388 100644 --- a/logd/src/logd/legacy_forwarder.cpp +++ b/logd/src/logd/legacy_forwarder.cpp @@ -58,26 +58,24 @@ LegacyForwarder::sendMode() } void -LegacyForwarder::forwardLine(const char *line, const char *eol) +LegacyForwarder::forwardLine(std::string_view line) { - int linelen = eol - line; - assert(_logserverfd >= 0); - assert (linelen > 0); - assert (linelen < 1024*1024); - assert (line[linelen - 1] == '\n'); + assert (line.size() > 0); + assert (line.size() < 1024*1024); + assert (line[line.size() - 1] == '\n'); - if (parseline(line, eol)) { - forwardText(line, linelen); + if (parseLine(line)) { + forwardText(line.data(), line.size()); } } bool -LegacyForwarder::parseline(const char *linestart, const char *lineend) +LegacyForwarder::parseLine(std::string_view line) { LogMessage message; try { - message.parse_log_line(std::string_view(linestart, lineend - linestart)); + message.parse_log_line(line); } catch (BadLogLineException &e) { LOG(spam, "bad logline: %s", e.what()); ++_badLines; diff --git a/logd/src/logd/legacy_forwarder.h b/logd/src/logd/legacy_forwarder.h index 81a93ce1d50..db3bf84fd4f 100644 --- a/logd/src/logd/legacy_forwarder.h +++ b/logd/src/logd/legacy_forwarder.h @@ -2,15 +2,9 @@ #pragma once #include "forwarder.h" -#include <vespa/log/log.h> -#include <map> -#include <unordered_set> namespace logdemon { -// Mapping saying if a level should be forwarded or not -using ForwardMap = std::map<ns_log::Logger::LogLevel, bool>; - struct Metrics; /** @@ -29,12 +23,13 @@ private: ret[len] = '\0'; return ret; } - bool parseline(const char *linestart, const char *lineend); + bool parseLine(std::string_view line); public: LegacyForwarder(Metrics &metrics); ~LegacyForwarder(); void forwardText(const char *text, int len); - void forwardLine(const char *line, const char *eol) override; + void forwardLine(std::string_view line) override; + void flush() override {} void setForwardMap(const ForwardMap & forwardMap) { _forwardMap = forwardMap; } void setLogserverFD(int fd) { _logserverfd = fd; } int getLogserverFD() { return _logserverfd; } diff --git a/logd/src/logd/proto_converter.h b/logd/src/logd/proto_converter.h index 688648b99de..88749100736 100644 --- a/logd/src/logd/proto_converter.h +++ b/logd/src/logd/proto_converter.h @@ -1,5 +1,7 @@ // Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +#pragma once + #include "log_protocol_proto.h" #include <vespa/log/log_message.h> #include <vector> @@ -11,6 +13,7 @@ namespace logdemon { */ struct ProtoConverter { using ProtoLogRequest = logserver::protocol::protobuf::LogRequest; + using ProtoLogResponse = logserver::protocol::protobuf::LogResponse; using ProtoLogMessage = logserver::protocol::protobuf::LogMessage; static void log_messages_to_proto(const std::vector<ns_log::LogMessage>& messages, ProtoLogRequest& proto); diff --git a/logd/src/logd/rpc_forwarder.cpp b/logd/src/logd/rpc_forwarder.cpp new file mode 100644 index 00000000000..e515f463db4 --- /dev/null +++ b/logd/src/logd/rpc_forwarder.cpp @@ -0,0 +1,151 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include "exceptions.h" +#include "metrics.h" +#include "proto_converter.h" +#include "rpc_forwarder.h" +#include <vespa/log/exceptions.h> +#include <vespa/vespalib/util/buffer.h> +#include <vespa/vespalib/util/stringfmt.h> + +#include <vespa/log/log.h> +LOG_SETUP(".logd.rpc_forwarder"); + +using ns_log::BadLogLineException; +using ns_log::LogMessage; +using vespalib::make_string; + +namespace logdemon { + +RpcForwarder::RpcForwarder(Metrics& metrics, FRT_Supervisor& supervisor, + const vespalib::string &hostname, int rpc_port, + double rpc_timeout_secs, size_t max_messages_per_request) + : _metrics(metrics), + _connection_spec(make_string("tcp/%s:%d", hostname.c_str(), rpc_port)), + _rpc_timeout_secs(rpc_timeout_secs), + _max_messages_per_request(max_messages_per_request), + _target(), + _messages(), + _bad_lines(0), + _forward_filter() +{ + _target = supervisor.GetTarget(_connection_spec.c_str()); +} + +RpcForwarder::~RpcForwarder() +{ + _target->SubRef(); +} + +namespace { + +void +encode_log_request(const ProtoConverter::ProtoLogRequest& src, FRT_RPCRequest& dst) +{ + dst.SetMethodName("vespa.logserver.archiveLogMessages"); + auto buf = src.SerializeAsString(); + auto& params = *dst.GetParams(); + params.AddInt8(0); // '0' indicates no compression + params.AddInt32(buf.size()); + params.AddData(buf.data(), buf.size()); +} + +bool +decode_log_response(FRT_RPCRequest& src, ProtoConverter::ProtoLogResponse& dst) +{ + auto& values = *src.GetReturn(); + uint8_t encoding = values[0]._intval8; + assert(encoding == 0); // Not using compression + uint32_t uncompressed_size = values[1]._intval32; + (void) uncompressed_size; + return dst.ParseFromArray(values[2]._data._buf, values[2]._data._len); +} + +bool +should_forward_log_message(const LogMessage& message, const ForwardMap& filter) +{ + auto found = filter.find(message.level()); + if (found != filter.end()) { + return found->second; + } + return false; +} + +} + +void +RpcForwarder::forwardLine(std::string_view line) +{ + LogMessage message; + try { + message.parse_log_line(line); + } catch (BadLogLineException &e) { + LOG(spam, "Skipping bad logline: %s", e.what()); + ++_bad_lines; + return; + } + _metrics.countLine(ns_log::Logger::logLevelNames[message.level()], message.service()); + if (should_forward_log_message(message, _forward_filter)) { + _messages.push_back(std::move(message)); + if (_messages.size() == _max_messages_per_request) { + flush(); + } + } +} + +namespace { + +class GuardedRequest { +private: + FRT_RPCRequest* _request; +public: + GuardedRequest() + : _request(new FRT_RPCRequest()) + {} + ~GuardedRequest() { + _request->SubRef(); + } + FRT_RPCRequest& operator*() const { return *_request; } + FRT_RPCRequest* get() const { return _request; } + FRT_RPCRequest* operator->() const { return get(); } +}; + +} + +void +RpcForwarder::flush() +{ + if (_messages.empty()) { + return; + } + ProtoConverter::ProtoLogRequest proto_request; + ProtoConverter::log_messages_to_proto(_messages, proto_request); + GuardedRequest request; + encode_log_request(proto_request, *request); + _target->InvokeSync(request.get(), _rpc_timeout_secs); + if (!request->CheckReturnTypes("bix")) { + auto error_msg = make_string("Error in rpc reply from '%s': '%s'", + _connection_spec.c_str(), request->GetErrorMessage()); + throw ConnectionException(error_msg); + } + ProtoConverter::ProtoLogResponse proto_response; + if (!decode_log_response(*request, proto_response)) { + auto error_msg = make_string("Error during decoding of protobuf response from '%s'", _connection_spec.c_str()); + throw DecodeException(error_msg); + } + _messages.clear(); +} + +int +RpcForwarder::badLines() const +{ + return _bad_lines; +} + +void +RpcForwarder::resetBadLines() +{ + _bad_lines = 0; +} + +} diff --git a/logd/src/logd/rpc_forwarder.h b/logd/src/logd/rpc_forwarder.h new file mode 100644 index 00000000000..3212da08195 --- /dev/null +++ b/logd/src/logd/rpc_forwarder.h @@ -0,0 +1,45 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#pragma once + +#include "forwarder.h" +#include "proto_converter.h" +#include <vespa/log/log_message.h> +#include <vespa/fnet/frt/frt.h> +#include <vector> + +namespace logdemon { + +struct Metrics; + +/** + * Implementation of the Forwarder interface that uses RPC to send protobuf encoded log messages to the logserver. + */ +class RpcForwarder : public Forwarder { +private: + Metrics& _metrics; + vespalib::string _connection_spec; + double _rpc_timeout_secs; + size_t _max_messages_per_request; + FRT_Target* _target; + std::vector<ns_log::LogMessage> _messages; + int _bad_lines; + ForwardMap _forward_filter; + +public: + RpcForwarder(Metrics& metrics, FRT_Supervisor& supervisor, + const vespalib::string& logserver_host, int logserver_rpc_port, + double rpc_timeout_secs, size_t max_messages_per_request); + ~RpcForwarder() override; + void set_forward_filter(const ForwardMap& forward_filter) { _forward_filter = forward_filter; } + + // Implements Forwarder + void sendMode() override {} + void forwardLine(std::string_view line) override; + void flush() override; + int badLines() const override; + void resetBadLines() override; +}; + +} + diff --git a/logd/src/logd/watcher.cpp b/logd/src/logd/watcher.cpp index a047c110f32..c505d2dd235 100644 --- a/logd/src/logd/watcher.cpp +++ b/logd/src/logd/watcher.cpp @@ -220,7 +220,7 @@ Watcher::watchfile() } while (nnl != nullptr && elapsed(tickStart) < 1) { ++nnl; - _forwarder.forwardLine(l, nnl); + _forwarder.forwardLine(std::string_view(l, (nnl - l))); ssize_t wsize = nnl - l; offset += wsize; l = nnl; diff --git a/logd/src/tests/legacy_forwarder/legacy_forwarder_test.cpp b/logd/src/tests/legacy_forwarder/legacy_forwarder_test.cpp index 3af35f9aa09..c6702e8bc67 100644 --- a/logd/src/tests/legacy_forwarder/legacy_forwarder_test.cpp +++ b/logd/src/tests/legacy_forwarder/legacy_forwarder_test.cpp @@ -40,13 +40,12 @@ struct ForwardFixture { } void verifyForward(bool doForward) { - const std::string & line(logLine); - forwarder.forwardLine(line.c_str(), line.c_str() + line.length()); + forwarder.forwardLine(logLine); fsync(fd); int rfd = open(fname.c_str(), O_RDONLY); char *buffer[2048]; ssize_t bytes = read(rfd, buffer, 2048); - ssize_t expected = doForward ? line.length() : 0; + ssize_t expected = doForward ? logLine.length() : 0; EXPECT_EQUAL(expected, bytes); close(rfd); } diff --git a/logd/src/tests/rpc_forwarder/CMakeLists.txt b/logd/src/tests/rpc_forwarder/CMakeLists.txt new file mode 100644 index 00000000000..66a30777b41 --- /dev/null +++ b/logd/src/tests/rpc_forwarder/CMakeLists.txt @@ -0,0 +1,9 @@ +# Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +vespa_add_executable(logd_rpc_forwarder_test_app TEST + SOURCES + rpc_forwarder_test.cpp + DEPENDS + logd + gtest +) +vespa_add_test(NAME logd_rpc_forwarder_test_app COMMAND logd_rpc_forwarder_test_app) diff --git a/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp new file mode 100644 index 00000000000..30ca5e19d44 --- /dev/null +++ b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp @@ -0,0 +1,248 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include <logd/exceptions.h> +#include <logd/metrics.h> +#include <logd/rpc_forwarder.h> +#include <vespa/vespalib/gtest/gtest.h> +#include <vespa/vespalib/metrics/dummy_metrics_manager.h> + +using namespace logdemon; +using vespalib::metrics::DummyMetricsManager; + +void +encode_log_response(const ProtoConverter::ProtoLogResponse& src, FRT_Values& dst) +{ + auto buf = src.SerializeAsString(); + dst.AddInt8(0); + dst.AddInt32(buf.size()); + dst.AddData(buf.data(), buf.size()); +} + +bool +decode_log_request(FRT_Values& src, ProtoConverter::ProtoLogRequest& dst) +{ + uint8_t encoding = src[0]._intval8; + assert(encoding == 0); + uint32_t uncompressed_size = src[1]._intval32; + assert(uncompressed_size == src[2]._data._len); + return dst.ParseFromArray(src[2]._data._buf, src[2]._data._len); +} + +std::string garbage("garbage"); + +struct RpcServer : public FRT_Invokable { + FRT_Supervisor supervisor; + int request_count; + std::vector<std::string> messages; + bool reply_with_error; + bool reply_with_proto_response; + +public: + RpcServer() + : supervisor(), + request_count(0), + messages(), + reply_with_error(false), + reply_with_proto_response(true) + { + supervisor.Listen(0); + FRT_ReflectionBuilder builder(&supervisor); + builder.DefineMethod("vespa.logserver.archiveLogMessages", "bix", "bix", + FRT_METHOD(RpcServer::rpc_archive_log_messages), this); + supervisor.Start(); + } + ~RpcServer() { + supervisor.ShutDown(true); + } + int get_listen_port() { + return supervisor.GetListenPort(); + } + void rpc_archive_log_messages(FRT_RPCRequest* request) { + ProtoConverter::ProtoLogRequest proto_request; + ASSERT_TRUE(decode_log_request(*request->GetParams(), proto_request)); + ++request_count; + for (const auto& message : proto_request.log_messages()) { + messages.push_back(message.payload()); + } + if (reply_with_error) { + request->SetError(123, "This is a server error"); + return; + } + if (reply_with_proto_response) { + ProtoConverter::ProtoLogResponse proto_response; + encode_log_response(proto_response, *request->GetReturn()); + } else { + auto& dst = *request->GetReturn(); + dst.AddInt8(0); + dst.AddInt32(garbage.size()); + dst.AddData(garbage.data(), garbage.size()); + } + } +}; + +std::string +make_log_line(const std::string& level, const std::string& payload) +{ + return "1234.5678\tmy_host\t10/20\tmy_service\tmy_component\t" + level + "\t" + payload; +} + +struct MockMetricsManager : public DummyMetricsManager { + int add_count; + MockMetricsManager() : DummyMetricsManager(), add_count(0) {} + void add(Counter::Increment) override { + ++add_count; + } +}; + +class ClientSupervisor { +private: + FRT_Supervisor _supervisor; +public: + ClientSupervisor() + : _supervisor() + { + _supervisor.Start(); + } + ~ClientSupervisor() { + _supervisor.ShutDown(true); + } + FRT_Supervisor& get() { return _supervisor; } + +}; + +struct RpcForwarderTest : public ::testing::Test { + RpcServer server; + std::shared_ptr<MockMetricsManager> metrics_mgr; + Metrics metrics; + ClientSupervisor supervisor; + RpcForwarder forwarder; + RpcForwarderTest() + : server(), + metrics_mgr(std::make_shared<MockMetricsManager>()), + metrics(metrics_mgr), + forwarder(metrics, supervisor.get(), "localhost", server.get_listen_port(), 60.0, 3) + { + ForwardMap forward_filter; + forward_filter[ns_log::Logger::error] = true; + forward_filter[ns_log::Logger::warning] = false; + forward_filter[ns_log::Logger::info] = true; + // all other log levels are implicit false + forwarder.set_forward_filter(forward_filter); + } + void forward_line(const std::string& payload) { + forwarder.forwardLine(make_log_line("info", payload)); + } + void forward_line(const std::string& level, const std::string& payload) { + forwarder.forwardLine(make_log_line(level, payload)); + } + void forward_bad_line() { + forwarder.forwardLine("badline"); + } + void flush() { + forwarder.flush(); + } + void expect_messages() { + expect_messages(0, {}); + } + void expect_messages(int exp_request_count, const std::vector<std::string>& exp_messages) { + EXPECT_EQ(exp_request_count, server.request_count); + EXPECT_EQ(exp_messages, server.messages); + } +}; + +TEST_F(RpcForwarderTest, does_not_send_rpc_with_no_log_messages) +{ + expect_messages(); + flush(); + expect_messages(); +} + +TEST_F(RpcForwarderTest, can_send_rpc_with_single_log_message) +{ + forward_line("a"); + expect_messages(); + flush(); + expect_messages(1, {"a"}); +} + +TEST_F(RpcForwarderTest, can_send_rpc_with_multiple_log_messages) +{ + forward_line("a"); + forward_line("b"); + expect_messages(); + flush(); + expect_messages(1, {"a", "b"}); +} + +TEST_F(RpcForwarderTest, automatically_sends_rpc_when_max_messages_limit_is_reached) +{ + forward_line("a"); + forward_line("b"); + expect_messages(); + forward_line("c"); + expect_messages(1, {"a", "b", "c"}); + forward_line("d"); + expect_messages(1, {"a", "b", "c"}); + forward_line("e"); + expect_messages(1, {"a", "b", "c"}); + forward_line("f"); + expect_messages(2, {"a", "b", "c", "d", "e", "f"}); +} + +TEST_F(RpcForwarderTest, bad_log_lines_are_counted_but_not_sent) +{ + forward_line("a"); + forward_bad_line(); + EXPECT_EQ(1, forwarder.badLines()); + flush(); + expect_messages(1, {"a"}); +} + +TEST_F(RpcForwarderTest, bad_log_lines_count_can_be_reset) +{ + forward_bad_line(); + EXPECT_EQ(1, forwarder.badLines()); + forwarder.resetBadLines(); + EXPECT_EQ(0, forwarder.badLines()); +} + +TEST_F(RpcForwarderTest, metrics_are_updated_for_each_log_message) +{ + forward_line("a"); + EXPECT_EQ(1, metrics_mgr->add_count); + forward_line("b"); + EXPECT_EQ(2, metrics_mgr->add_count); +} + +TEST_F(RpcForwarderTest, log_messages_are_filtered_on_log_level) +{ + forward_line("fatal", "a"); + forward_line("error", "b"); + forward_line("warning", "c"); + forward_line("config", "d"); + forward_line("info", "e"); + forward_line("event", "f"); + forward_line("debug", "g"); + forward_line("spam", "h"); + forward_line("null", "i"); + flush(); + expect_messages(1, {"b", "e"}); + EXPECT_EQ(9, metrics_mgr->add_count); +} + +TEST_F(RpcForwarderTest, throws_when_rpc_reply_contains_errors) +{ + server.reply_with_error = true; + forward_line("a"); + EXPECT_THROW(flush(), logdemon::ConnectionException); +} + +TEST_F(RpcForwarderTest, throws_when_rpc_reply_does_not_contain_proto_response) +{ + server.reply_with_proto_response = false; + forward_line("a"); + EXPECT_THROW(flush(), logdemon::DecodeException); +} + +GTEST_MAIN_RUN_ALL_TESTS() + diff --git a/metrics/src/tests/CMakeLists.txt b/metrics/src/tests/CMakeLists.txt index afb56757bdd..2c69b83d4b7 100644 --- a/metrics/src/tests/CMakeLists.txt +++ b/metrics/src/tests/CMakeLists.txt @@ -38,6 +38,5 @@ vespa_add_executable(metrics_testrunner_app TEST # TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME metrics_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:metrics_testrunner_app> - DEPENDS metrics_testrunner_app + COMMAND metrics_testrunner_app ) diff --git a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/AthenzCredentialsMaintainer.java b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/AthenzCredentialsMaintainer.java index abac16d242e..4fe0f420f05 100644 --- a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/AthenzCredentialsMaintainer.java +++ b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/AthenzCredentialsMaintainer.java @@ -17,8 +17,8 @@ import com.yahoo.vespa.athenz.identity.ServiceIdentityProvider; import com.yahoo.vespa.athenz.identityprovider.api.EntityBindingsMapper; import com.yahoo.vespa.athenz.identityprovider.api.IdentityDocumentClient; import com.yahoo.vespa.athenz.identityprovider.api.SignedIdentityDocument; -import com.yahoo.vespa.athenz.identityprovider.client.DefaultIdentityDocumentClient; import com.yahoo.vespa.athenz.identityprovider.client.CsrGenerator; +import com.yahoo.vespa.athenz.identityprovider.client.DefaultIdentityDocumentClient; import com.yahoo.vespa.athenz.tls.AthenzIdentityVerifier; import com.yahoo.vespa.athenz.utils.SiaUtils; import com.yahoo.vespa.hosted.dockerapi.ContainerName; @@ -51,7 +51,7 @@ import static java.util.Collections.singleton; * * @author bjorncs */ -public class AthenzCredentialsMaintainer { +public class AthenzCredentialsMaintainer implements CredentialsMaintainer { private static final Logger logger = Logger.getLogger(AthenzCredentialsMaintainer.class.getName()); @@ -89,7 +89,7 @@ public class AthenzCredentialsMaintainer { this.clock = Clock.systemUTC(); } - public void converge(NodeAgentContext context) { + public boolean converge(NodeAgentContext context) { try { context.log(logger, LogLevel.DEBUG, "Checking certificate"); Path containerSiaDirectory = context.pathOnHostFromPathInNode(CONTAINER_SIA_DIRECTORY); @@ -102,7 +102,7 @@ public class AthenzCredentialsMaintainer { Files.createDirectories(certificateFile.getParent()); Files.createDirectories(identityDocumentFile.getParent()); registerIdentity(context, privateKeyFile, certificateFile, identityDocumentFile); - return; + return true; } X509Certificate certificate = readCertificateFromFile(certificateFile); @@ -111,7 +111,7 @@ public class AthenzCredentialsMaintainer { if (isCertificateExpired(expiry, now)) { context.log(logger, "Certificate has expired (expiry=%s)", expiry.toString()); registerIdentity(context, privateKeyFile, certificateFile, identityDocumentFile); - return; + return true; } Duration age = Duration.between(certificate.getNotBefore().toInstant(), now); @@ -121,14 +121,15 @@ public class AthenzCredentialsMaintainer { context.log(logger, LogLevel.WARNING, String.format( "Skipping refresh attempt as last refresh was on %s (less than %s ago)", lastRefreshAttempt.get(context.containerName()).toString(), REFRESH_BACKOFF.toString())); - return; + return false; } else { lastRefreshAttempt.put(context.containerName(), now); refreshIdentity(context, privateKeyFile, certificateFile, identityDocumentFile); - return; + return true; } } context.log(logger, LogLevel.DEBUG, "Certificate is still valid"); + return false; } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/CredentialsMaintainer.java b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/CredentialsMaintainer.java new file mode 100644 index 00000000000..58c3585a48f --- /dev/null +++ b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/maintenance/identity/CredentialsMaintainer.java @@ -0,0 +1,21 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +package com.yahoo.vespa.hosted.node.admin.maintenance.identity; + +import com.yahoo.vespa.hosted.node.admin.nodeagent.NodeAgentContext; + +/** + * A maintainer that is responsible for providing and refreshing credentials for a container. + * + * @author freva + */ +public interface CredentialsMaintainer { + + /** + * Creates/refreshes credentials for the given NodeAgentContext. Called for every NodeAgent tick. + * @return false if already converged, i.e. was a no-op. + */ + boolean converge(NodeAgentContext context); + + /** Remove any existing credentials. This method is called just before container data is archived. */ + void clearCredentials(NodeAgentContext context); +} diff --git a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImpl.java b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImpl.java index fc4162a0f73..57e504a6ffd 100644 --- a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImpl.java +++ b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImpl.java @@ -27,7 +27,7 @@ import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.OrchestratorE import com.yahoo.vespa.hosted.node.admin.docker.DockerOperations; import com.yahoo.vespa.hosted.node.admin.maintenance.StorageMaintainer; import com.yahoo.vespa.hosted.node.admin.maintenance.acl.AclMaintainer; -import com.yahoo.vespa.hosted.node.admin.maintenance.identity.AthenzCredentialsMaintainer; +import com.yahoo.vespa.hosted.node.admin.maintenance.identity.CredentialsMaintainer; import com.yahoo.vespa.hosted.node.admin.nodeadmin.ConvergenceException; import com.yahoo.vespa.hosted.node.admin.util.SecretAgentCheckConfig; @@ -63,7 +63,7 @@ public class NodeAgentImpl implements NodeAgent { private final Orchestrator orchestrator; private final DockerOperations dockerOperations; private final StorageMaintainer storageMaintainer; - private final Optional<AthenzCredentialsMaintainer> athenzCredentialsMaintainer; + private final Optional<CredentialsMaintainer> credentialsMaintainer; private final Optional<AclMaintainer> aclMaintainer; private final Optional<HealthChecker> healthChecker; @@ -105,7 +105,7 @@ public class NodeAgentImpl implements NodeAgent { final DockerOperations dockerOperations, final StorageMaintainer storageMaintainer, final FlagSource flagSource, - final Optional<AthenzCredentialsMaintainer> athenzCredentialsMaintainer, + final Optional<CredentialsMaintainer> credentialsMaintainer, final Optional<AclMaintainer> aclMaintainer, final Optional<HealthChecker> healthChecker) { this.contextSupplier = contextSupplier; @@ -113,7 +113,7 @@ public class NodeAgentImpl implements NodeAgent { this.orchestrator = orchestrator; this.dockerOperations = dockerOperations; this.storageMaintainer = storageMaintainer; - this.athenzCredentialsMaintainer = athenzCredentialsMaintainer; + this.credentialsMaintainer = credentialsMaintainer; this.aclMaintainer = aclMaintainer; this.healthChecker = healthChecker; @@ -443,7 +443,7 @@ public class NodeAgentImpl implements NodeAgent { return; } container = removeContainerIfNeededUpdateContainerState(context, container); - athenzCredentialsMaintainer.ifPresent(maintainer -> maintainer.converge(context)); + credentialsMaintainer.ifPresent(maintainer -> maintainer.converge(context)); if (! container.isPresent()) { containerState = STARTING; startContainer(context); @@ -481,7 +481,7 @@ public class NodeAgentImpl implements NodeAgent { case dirty: removeContainerIfNeededUpdateContainerState(context, container); context.log(logger, "State is " + node.getState() + ", will delete application storage and mark node as ready"); - athenzCredentialsMaintainer.ifPresent(maintainer -> maintainer.clearCredentials(context)); + credentialsMaintainer.ifPresent(maintainer -> maintainer.clearCredentials(context)); storageMaintainer.archiveNodeStorage(context); updateNodeRepoWithCurrentAttributes(context); nodeRepository.setNodeState(context.hostname().value(), NodeState.ready); diff --git a/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImplTest.java b/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImplTest.java index 15786f2ed34..dcbbaf792a8 100644 --- a/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImplTest.java +++ b/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentImplTest.java @@ -25,7 +25,7 @@ import com.yahoo.vespa.hosted.node.admin.configserver.orchestrator.OrchestratorE import com.yahoo.vespa.hosted.node.admin.docker.DockerOperations; import com.yahoo.vespa.hosted.node.admin.maintenance.StorageMaintainer; import com.yahoo.vespa.hosted.node.admin.maintenance.acl.AclMaintainer; -import com.yahoo.vespa.hosted.node.admin.maintenance.identity.AthenzCredentialsMaintainer; +import com.yahoo.vespa.hosted.node.admin.maintenance.identity.CredentialsMaintainer; import com.yahoo.vespa.hosted.node.admin.nodeadmin.ConvergenceException; import org.junit.Test; import org.mockito.InOrder; @@ -84,7 +84,7 @@ public class NodeAgentImplTest { private final HealthChecker healthChecker = mock(HealthChecker.class); private final ContainerStats emptyContainerStats = new ContainerStats(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap()); - private final AthenzCredentialsMaintainer athenzCredentialsMaintainer = mock(AthenzCredentialsMaintainer.class); + private final CredentialsMaintainer credentialsMaintainer = mock(CredentialsMaintainer.class); private final InMemoryFlagSource flagSource = new InMemoryFlagSource(); @@ -763,7 +763,7 @@ public class NodeAgentImplTest { doNothing().when(storageMaintainer).writeMetricsConfig(any()); return new NodeAgentImpl(contextSupplier, nodeRepository, orchestrator, dockerOperations, - storageMaintainer, flagSource, Optional.of(athenzCredentialsMaintainer), Optional.of(aclMaintainer), + storageMaintainer, flagSource, Optional.of(credentialsMaintainer), Optional.of(aclMaintainer), Optional.of(healthChecker)); } diff --git a/persistence/src/tests/CMakeLists.txt b/persistence/src/tests/CMakeLists.txt index d6f6db07f9d..396e9f99655 100644 --- a/persistence/src/tests/CMakeLists.txt +++ b/persistence/src/tests/CMakeLists.txt @@ -7,9 +7,7 @@ vespa_add_executable(persistence_testrunner_app TEST persistence_testspi ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME persistence_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:persistence_testrunner_app> - DEPENDS persistence_testrunner_app + COMMAND persistence_testrunner_app ) diff --git a/searchlib/src/vespa/searchlib/attribute/attribute_blueprint_factory.cpp b/searchlib/src/vespa/searchlib/attribute/attribute_blueprint_factory.cpp index f09f48363db..07689e5ffec 100644 --- a/searchlib/src/vespa/searchlib/attribute/attribute_blueprint_factory.cpp +++ b/searchlib/src/vespa/searchlib/attribute/attribute_blueprint_factory.cpp @@ -115,6 +115,13 @@ public: return _search_context->createIterator(tfmda[0], strict); } + SearchIterator::UP + createSearch(fef::MatchData &md, bool strict) const override { + const State &state = getState(); + assert(state.numFields() == 1); + return _search_context->createIterator(state.field(0).resolve(md), strict); + } + void fetchPostings(bool strict) override { _search_context->fetchPostings(strict); diff --git a/searchlib/src/vespa/searchlib/queryeval/weighted_set_term_blueprint.cpp b/searchlib/src/vespa/searchlib/queryeval/weighted_set_term_blueprint.cpp index fc68c48a247..cec72129475 100644 --- a/searchlib/src/vespa/searchlib/queryeval/weighted_set_term_blueprint.cpp +++ b/searchlib/src/vespa/searchlib/queryeval/weighted_set_term_blueprint.cpp @@ -44,8 +44,7 @@ WeightedSetTermBlueprint::addTerm(Blueprint::UP term, int32_t weight) SearchIterator::UP -WeightedSetTermBlueprint::createLeafSearch(const search::fef::TermFieldMatchDataArray &tfmda, - bool) const +WeightedSetTermBlueprint::createLeafSearch(const fef::TermFieldMatchDataArray &tfmda, bool) const { assert(tfmda.size() == 1); fef::MatchData::UP md = _layout.createMatchData(); diff --git a/staging_vespalib/src/vespa/vespalib/metrics/dummy_metrics_manager.h b/staging_vespalib/src/vespa/vespalib/metrics/dummy_metrics_manager.h index e4ca1e0ec49..6aeb1137732 100644 --- a/staging_vespalib/src/vespa/vespalib/metrics/dummy_metrics_manager.h +++ b/staging_vespalib/src/vespa/vespalib/metrics/dummy_metrics_manager.h @@ -20,7 +20,7 @@ namespace metrics { **/ class DummyMetricsManager : public MetricsManager { -private: +protected: DummyMetricsManager() {} public: ~DummyMetricsManager(); diff --git a/storageapi/src/tests/CMakeLists.txt b/storageapi/src/tests/CMakeLists.txt index 2045cca89e1..ebbf3b8357a 100644 --- a/storageapi/src/tests/CMakeLists.txt +++ b/storageapi/src/tests/CMakeLists.txt @@ -26,9 +26,7 @@ vespa_add_executable(storageapi_testrunner_app TEST storageapi ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME storageapi_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:storageapi_testrunner_app> - DEPENDS storageapi_testrunner_app + COMMAND storageapi_testrunner_app ) diff --git a/storageframework/src/tests/CMakeLists.txt b/storageframework/src/tests/CMakeLists.txt index 49c25a275f2..aa4be0783fa 100644 --- a/storageframework/src/tests/CMakeLists.txt +++ b/storageframework/src/tests/CMakeLists.txt @@ -24,9 +24,7 @@ vespa_add_executable(storageframework_testrunner_app TEST storageframework_testthread ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME storageframework_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:storageframework_testrunner_app> - DEPENDS storageframework_testrunner_app + COMMAND storageframework_testrunner_app ) diff --git a/storageserver/src/tests/CMakeLists.txt b/storageserver/src/tests/CMakeLists.txt index a78aea8aea3..21412e4bc33 100644 --- a/storageserver/src/tests/CMakeLists.txt +++ b/storageserver/src/tests/CMakeLists.txt @@ -11,9 +11,7 @@ vespa_add_executable(storageserver_testrunner_app TEST searchlib_searchlib_uca ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME storageserver_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:storageserver_testrunner_app> - DEPENDS storageserver_testrunner_app + COMMAND storageserver_testrunner_app ) diff --git a/vdslib/src/tests/CMakeLists.txt b/vdslib/src/tests/CMakeLists.txt index e1853ab05bd..1fdc45a3163 100644 --- a/vdslib/src/tests/CMakeLists.txt +++ b/vdslib/src/tests/CMakeLists.txt @@ -29,6 +29,5 @@ vespa_add_executable(vdslib_testrunner_app TEST # TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME vdslib_testrunner_app - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:vdslib_testrunner_app> - DEPENDS vdslib_testrunner_app + COMMAND vdslib_testrunner_app ) diff --git a/vdstestlib/src/tests/cppunit/CMakeLists.txt b/vdstestlib/src/tests/cppunit/CMakeLists.txt index 5cb577ce70e..1b8e857fe30 100644 --- a/vdstestlib/src/tests/cppunit/CMakeLists.txt +++ b/vdstestlib/src/tests/cppunit/CMakeLists.txt @@ -7,10 +7,8 @@ vespa_add_executable(vdstestlib_testrunner_app TEST vdstestlib ) -# TODO: Test with a larger chunk size to parallelize test suite runs vespa_add_test( NAME vdstestlib_testrunner_app NO_VALGRIND - COMMAND python ${PROJECT_SOURCE_DIR}/cppunit-parallelize.py --chunks 1 $<TARGET_FILE:vdstestlib_testrunner_app> - DEPENDS vdstestlib_testrunner_app + COMMAND vdstestlib_testrunner_app ) diff --git a/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java b/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java index 6f5fb591ba6..24d4cfe4318 100644 --- a/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java +++ b/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java @@ -1,8 +1,6 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.yolean.concurrent; -import com.yahoo.yolean.concurrent.ResourceFactory; - import java.util.Iterator; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; diff --git a/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java index 62d5d749604..40c5ca3b6c2 100644 --- a/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java +++ b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java @@ -1,8 +1,6 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.yolean.concurrent; -import com.yahoo.yolean.concurrent.ResourceFactory; - import java.util.ArrayDeque; import java.util.Deque; import java.util.Iterator; |