You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aurora.apache.org by ma...@apache.org on 2016/04/26 01:19:47 UTC
[2/2] aurora git commit: Schema changes for resource management
refactoring
Schema changes for resource management refactoring
Reviewed at https://reviews.apache.org/r/46459/
Project: http://git-wip-us.apache.org/repos/asf/aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/ff6e05f0
Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/ff6e05f0
Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/ff6e05f0
Branch: refs/heads/master
Commit: ff6e05f058cb2191f405d1f691790b2c79b14f2b
Parents: d339036
Author: Maxim Khutornenko <ma...@apache.org>
Authored: Mon Apr 25 16:19:29 2016 -0700
Committer: Maxim Khutornenko <ma...@apache.org>
Committed: Mon Apr 25 16:19:29 2016 -0700
----------------------------------------------------------------------
.../thrift/org/apache/aurora/gen/api.thrift | 10 ++
.../aurora/scheduler/base/TaskTestUtil.java | 8 +-
.../configuration/ConfigurationManager.java | 17 +++
.../http/api/GsonMessageBodyHandler.java | 26 +++-
.../scheduler/resources/ResourceType.java | 140 +++++++++++++++++--
.../resources/ResourceTypeConverter.java | 65 +++++++++
.../aurora/scheduler/storage/db/DbStorage.java | 5 +
.../scheduler/storage/db/TaskConfigManager.java | 13 ++
.../scheduler/storage/db/TaskConfigMapper.java | 10 ++
.../V003_CreateResourceTypesTable.java | 56 ++++++++
.../migration/V004_CreateTaskResourceTable.java | 74 ++++++++++
.../db/typehandlers/ResourceTypeHandler.java | 26 ++++
.../storage/db/typehandlers/TypeHandlers.java | 1 +
.../scheduler/storage/db/views/DBResource.java | 30 ++++
.../storage/db/views/DbTaskConfig.java | 9 +-
.../scheduler/storage/log/LogStorage.java | 16 +--
.../storage/log/SnapshotStoreImpl.java | 9 +-
.../scheduler/storage/log/ThriftBackfill.java | 127 +++++++++++++++++
src/main/python/apache/aurora/config/thrift.py | 5 +
.../aurora/tools/java/thrift_wrapper_codegen.py | 57 ++++++--
.../scheduler/storage/db/TaskConfigMapper.xml | 36 +++++
.../aurora/scheduler/storage/db/schema.sql | 16 +++
.../aurora/scheduler/app/SchedulerIT.java | 16 ++-
.../configuration/ConfigurationManagerTest.java | 40 +++++-
.../resources/ResourceTypeConverterTest.java | 30 ++++
.../scheduler/resources/ResourceTypeTest.java | 36 +++++
.../scheduler/storage/backup/RecoveryTest.java | 6 +-
.../scheduler/storage/log/LogStorageTest.java | 41 ++++--
.../storage/log/SnapshotStoreImplIT.java | 24 ++++
.../storage/log/ThriftBackfillTest.java | 96 +++++++++++++
.../aurora/scheduler/thrift/Fixtures.java | 7 +-
.../thrift/ReadOnlySchedulerImplTest.java | 20 +--
.../thrift/SchedulerThriftInterfaceTest.java | 8 +-
.../apache/aurora/client/cli/test_inspect.py | 5 +-
.../python/apache/aurora/config/test_thrift.py | 10 +-
35 files changed, 1020 insertions(+), 75 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/api/src/main/thrift/org/apache/aurora/gen/api.thrift
----------------------------------------------------------------------
diff --git a/api/src/main/thrift/org/apache/aurora/gen/api.thrift b/api/src/main/thrift/org/apache/aurora/gen/api.thrift
index 0cf4d6e..b1b2f1d 100644
--- a/api/src/main/thrift/org/apache/aurora/gen/api.thrift
+++ b/api/src/main/thrift/org/apache/aurora/gen/api.thrift
@@ -224,6 +224,14 @@ union Image {
2: AppcImage appc
}
+/** Describes resource value required to run a task. */
+union Resource {
+ 1: double numCpus
+ 2: i64 ramMb
+ 3: i64 diskMb
+ 4: string namedPort
+}
+
/** Description of the tasks contained within a job. */
struct TaskConfig {
/** Job task belongs to. */
@@ -246,6 +254,8 @@ struct TaskConfig {
* specifying a container)
*/
31: optional Image image
+ /** All resources required to run a task. */
+ 32: set<Resource> resources
20: set<Constraint> constraints
/** a list of named ports this task requests */
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/base/TaskTestUtil.java b/src/main/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
index d91c742..221417f 100644
--- a/src/main/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
+++ b/src/main/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
@@ -28,6 +28,7 @@ import org.apache.aurora.gen.ExecutorConfig;
import org.apache.aurora.gen.Identity;
import org.apache.aurora.gen.LimitConstraint;
import org.apache.aurora.gen.Metadata;
+import org.apache.aurora.gen.Resource;
import org.apache.aurora.gen.ScheduleStatus;
import org.apache.aurora.gen.ScheduledTask;
import org.apache.aurora.gen.TaskConfig;
@@ -104,7 +105,12 @@ public final class TaskTestUtil {
new DockerContainer("imagename")
.setParameters(ImmutableList.of(
new DockerParameter("a", "b"),
- new DockerParameter("c", "d"))))));
+ new DockerParameter("c", "d")))))
+ .setResources(ImmutableSet.of(
+ Resource.numCpus(1.0),
+ Resource.ramMb(1024),
+ Resource.diskMb(1024),
+ Resource.namedPort("http"))));
}
public static IScheduledTask makeTask(String id, IJobKey job) {
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java b/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
index 041cec3..9a15a4b 100644
--- a/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
+++ b/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
@@ -14,6 +14,7 @@
package org.apache.aurora.scheduler.configuration;
import java.util.Map;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.inject.Inject;
@@ -37,12 +38,14 @@ import org.apache.aurora.gen.TaskConstraint;
import org.apache.aurora.scheduler.TierManager;
import org.apache.aurora.scheduler.base.JobKeys;
import org.apache.aurora.scheduler.base.UserProvidedStrings;
+import org.apache.aurora.scheduler.resources.ResourceType;
import org.apache.aurora.scheduler.storage.entities.IConstraint;
import org.apache.aurora.scheduler.storage.entities.IContainer;
import org.apache.aurora.scheduler.storage.entities.IJobConfiguration;
import org.apache.aurora.scheduler.storage.entities.ITaskConfig;
import org.apache.aurora.scheduler.storage.entities.ITaskConstraint;
import org.apache.aurora.scheduler.storage.entities.IValueConstraint;
+import org.apache.aurora.scheduler.storage.log.ThriftBackfill;
import static java.util.Objects.requireNonNull;
@@ -310,6 +313,20 @@ public class ConfigurationManager {
throw new TaskDescriptionException(CONTAINER_AND_IMAGE_ARE_MUTUALLY_EXCLUSIVE);
}
+ ThriftBackfill.backfillTask(builder);
+
+ String types = config.getResources().stream()
+ .collect(Collectors.groupingBy(e -> ResourceType.fromResource(e)))
+ .entrySet().stream()
+ .filter(e -> !e.getKey().isMultipleAllowed() && e.getValue().size() > 1)
+ .map(r -> r.getKey().getAuroraName())
+ .sorted()
+ .collect(Collectors.joining(", "));
+
+ if (!Strings.isNullOrEmpty(types)) {
+ throw new TaskDescriptionException("Multiple resource values are not supported for " + types);
+ }
+
return ITaskConfig.build(builder);
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/http/api/GsonMessageBodyHandler.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/http/api/GsonMessageBodyHandler.java b/src/main/java/org/apache/aurora/scheduler/http/api/GsonMessageBodyHandler.java
index 44295f8..1007a32 100644
--- a/src/main/java/org/apache/aurora/scheduler/http/api/GsonMessageBodyHandler.java
+++ b/src/main/java/org/apache/aurora/scheduler/http/api/GsonMessageBodyHandler.java
@@ -52,7 +52,9 @@ import com.google.gson.JsonSerializer;
import org.apache.thrift.TFieldIdEnum;
import org.apache.thrift.TUnion;
import org.apache.thrift.meta_data.FieldMetaData;
+import org.apache.thrift.meta_data.FieldValueMetaData;
import org.apache.thrift.meta_data.StructMetaData;
+import org.apache.thrift.protocol.TType;
/**
* A message body reader/writer that uses gson to translate JSON to and from java objects produced
@@ -200,8 +202,28 @@ public class GsonMessageBodyHandler
for (Entry<TFieldIdEnum, FieldMetaData> entry : metaDataMap.entrySet()) {
if (entry.getKey().getFieldName().equals(item.getKey())) {
- StructMetaData valueMetaData = (StructMetaData) entry.getValue().valueMetaData;
- Object result = context.deserialize(item.getValue(), valueMetaData.structClass);
+ Object result;
+ if (entry.getValue().valueMetaData.isStruct()) {
+ StructMetaData valueMeta = (StructMetaData) entry.getValue().valueMetaData;
+ result = context.deserialize(item.getValue(), valueMeta.structClass);
+ } else {
+ FieldValueMetaData valueMeta = entry.getValue().valueMetaData;
+ Type type;
+ switch (valueMeta.type) {
+ case TType.DOUBLE:
+ type = Double.TYPE;
+ break;
+ case TType.I64:
+ type = Long.TYPE;
+ break;
+ case TType.STRING:
+ type = String.class;
+ break;
+ default:
+ throw new RuntimeException("Unmapped type: " + valueMeta.type);
+ }
+ result = context.deserialize(item.getValue(), type);
+ }
return createUnion(clazz, entry.getKey(), result);
}
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/resources/ResourceType.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/resources/ResourceType.java b/src/main/java/org/apache/aurora/scheduler/resources/ResourceType.java
index b6fc949..6e4d694 100644
--- a/src/main/java/org/apache/aurora/scheduler/resources/ResourceType.java
+++ b/src/main/java/org/apache/aurora/scheduler/resources/ResourceType.java
@@ -13,37 +13,62 @@
*/
package org.apache.aurora.scheduler.resources;
+import java.util.EnumSet;
+
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
import org.apache.aurora.common.quantity.Amount;
-import org.apache.aurora.common.quantity.Data;
+import org.apache.aurora.gen.Resource._Fields;
+import org.apache.aurora.scheduler.storage.entities.IResource;
+import org.apache.mesos.Protos;
+import org.apache.thrift.TEnum;
import static java.util.Objects.requireNonNull;
+import static org.apache.aurora.common.quantity.Data.GB;
+import static org.apache.aurora.common.quantity.Data.MB;
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.DOUBLE;
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.LONG;
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.STRING;
+import static org.apache.mesos.Protos.Value.Type.RANGES;
+import static org.apache.mesos.Protos.Value.Type.SCALAR;
+
/**
* Describes Mesos resource types and their Aurora traits.
*/
@VisibleForTesting
-public enum ResourceType {
+public enum ResourceType implements TEnum {
/**
* CPU resource.
*/
- CPUS("cpus", "CPU", 16),
+ CPUS(_Fields.NUM_CPUS, SCALAR, "cpus", DOUBLE, "CPU", 16, false),
/**
* RAM resource.
*/
- RAM_MB("mem", "RAM", Amount.of(24, Data.GB).as(Data.MB)),
+ RAM_MB(_Fields.RAM_MB, SCALAR, "mem", LONG, "RAM", Amount.of(24, GB).as(MB), false),
/**
* DISK resource.
*/
- DISK_MB("disk", "disk", Amount.of(450, Data.GB).as(Data.MB)),
+ DISK_MB(_Fields.DISK_MB, SCALAR, "disk", LONG, "disk", Amount.of(450, GB).as(MB), false),
/**
* Port resource.
*/
- PORTS("ports", "ports", 1000);
+ PORTS(_Fields.NAMED_PORT, RANGES, "ports", STRING, "ports", 1000, true);
+
+ /**
+ * Correspondent thrift {@link org.apache.aurora.gen.Resource} enum value.
+ */
+ private final _Fields value;
+
+ /**
+ * Mesos resource type.
+ */
+ private final Protos.Value.Type mesosType;
/**
* Mesos resource name.
@@ -51,28 +76,77 @@ public enum ResourceType {
private final String mesosName;
/**
+ * Type converter for resource values.
+ */
+ private final ResourceTypeConverter<?> typeConverter;
+
+ /**
* Aurora resource name.
*/
private final String auroraName;
/**
- * Scaling range to use for comparison of scheduling vetoes. This has no real bearing besides
- * trying to determine if a veto along one resource vector is a 'stronger' veto than that of
- * another vector. The value represents the typical slave machine resources.
+ * Scaling range for comparing scheduling vetoes.
*/
private final int scalingRange;
/**
+ * Indicates if multiple resource types are allowed in a task.
+ */
+ private final boolean isMultipleAllowed;
+
+ private static ImmutableMap<Integer, ResourceType> byField =
+ Maps.uniqueIndex(EnumSet.allOf(ResourceType.class), ResourceType::getValue);
+
+ /**
* Describes a Resource type.
*
+ * @param value Correspondent {@link _Fields} value.
+ * @param mesosType See {@link #getMesosType()} for more details.
* @param mesosName See {@link #getMesosName()} for more details.
+ * @param typeConverter See {@link #getTypeConverter()} for more details.
* @param auroraName See {@link #getAuroraName()} for more details.
* @param scalingRange See {@link #getScalingRange()} for more details.
+ * @param isMultipleAllowed See {@link #isMultipleAllowed()} for more details.
*/
- ResourceType(String mesosName, String auroraName, int scalingRange) {
+ ResourceType(
+ _Fields value,
+ Protos.Value.Type mesosType,
+ String mesosName,
+ ResourceTypeConverter<?> typeConverter,
+ String auroraName,
+ int scalingRange,
+ boolean isMultipleAllowed) {
+
+ this.value = value;
+ this.mesosType = requireNonNull(mesosType);
this.mesosName = requireNonNull(mesosName);
+ this.typeConverter = requireNonNull(typeConverter);
this.auroraName = requireNonNull(auroraName);
this.scalingRange = scalingRange;
+ this.isMultipleAllowed = isMultipleAllowed;
+ }
+
+ /**
+ * Get unique ID value.
+ *
+ * @return Enum ID.
+ */
+ @Override
+ public int getValue() {
+ return value.getThriftFieldId();
+ }
+
+ /**
+ * Gets Mesos resource type.
+ * <p>
+ * @see <a href="https://github.com/apache/mesos/blob/master/include/mesos/mesos.proto/">Mesos
+ * protobuf for more details</a>
+ *
+ * @return Mesos resource type.
+ */
+ public Protos.Value.Type getMesosType() {
+ return mesosType;
}
/**
@@ -88,6 +162,15 @@ public enum ResourceType {
}
/**
+ * Gets {@link ResourceTypeConverter} to convert resource values.
+ *
+ * @return {@link ResourceTypeConverter} instance.
+ */
+ public ResourceTypeConverter<?> getTypeConverter() {
+ return typeConverter;
+ }
+
+ /**
* Gets resource name for internal Aurora representation (e.g. in the UI).
*
* @return Aurora resource name.
@@ -97,11 +180,46 @@ public enum ResourceType {
}
/**
- * Returns scaling range for comparing scheduling vetoes.
+ * Scaling range to use for comparison of scheduling vetoes.
+ * <p>
+ * This has no real bearing besides trying to determine if a veto along one resource vector
+ * is a 'stronger' veto than that of another vector. The value represents the typical slave
+ * machine resources.
*
* @return Resource scaling range.
*/
public int getScalingRange() {
return scalingRange;
}
+
+ /**
+ * Returns a flag indicating if multiple resource of the same type are allowed in a given task.
+ *
+ * @return True if multiple resources of the same type are allowed, false otherwise.
+ */
+ public boolean isMultipleAllowed() {
+ return isMultipleAllowed;
+ }
+
+ /**
+ * Returns a {@link ResourceType} for the given ID.
+ *
+ * @param value ID value to search by. See {@link #getValue()}.
+ * @return {@link ResourceType}.
+ */
+ public static ResourceType fromIdValue(int value) {
+ return requireNonNull(byField.get(value), "Unmapped value: " + value);
+ }
+
+ /**
+ * Returns a {@link ResourceType} for the given resource.
+ *
+ * @param resource {@link IResource} to search by.
+ * @return {@link ResourceType}.
+ */
+ public static ResourceType fromResource(IResource resource) {
+ return requireNonNull(
+ byField.get((int) resource.getSetField().getThriftFieldId()),
+ "Unknown resource: " + resource);
+ }
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/resources/ResourceTypeConverter.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/resources/ResourceTypeConverter.java b/src/main/java/org/apache/aurora/scheduler/resources/ResourceTypeConverter.java
new file mode 100644
index 0000000..11395ec
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/resources/ResourceTypeConverter.java
@@ -0,0 +1,65 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.resources;
+
+import com.google.common.primitives.Longs;
+
+/**
+ * Converts resource values to/from generic (String) representation.
+ * @param <T> Resource value type to convert.
+ */
+public interface ResourceTypeConverter<T> {
+ /**
+ * Parses resource value from the string representation.
+ *
+ * @param value String value to parse.
+ * @return Resource value of type {@code T}.
+ */
+ T parseFrom(String value);
+
+ /**
+ * Converts resource of type {@code T} to its string representation.
+ *
+ * @param value Resource value to stringify.
+ * @return String representation of the resource value.
+ */
+ default String stringify(Object value) {
+ return value.toString();
+ }
+
+ LongConverter LONG = new LongConverter();
+ DoubleConverter DOUBLE = new DoubleConverter();
+ StringConverter STRING = new StringConverter();
+
+ class LongConverter implements ResourceTypeConverter<Long> {
+ @Override
+ public Long parseFrom(String value) {
+ return Longs.tryParse(value);
+ }
+ }
+
+ class DoubleConverter implements ResourceTypeConverter<Double> {
+ @Override
+ public Double parseFrom(String value) {
+ return Double.parseDouble(value);
+ }
+ }
+
+ class StringConverter implements ResourceTypeConverter<String> {
+ @Override
+ public String parseFrom(String value) {
+ return value;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/DbStorage.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/DbStorage.java b/src/main/java/org/apache/aurora/scheduler/storage/db/DbStorage.java
index 360914e..acb4498 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/db/DbStorage.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/DbStorage.java
@@ -34,6 +34,7 @@ import org.apache.aurora.gen.ScheduleStatus;
import org.apache.aurora.scheduler.async.AsyncModule.AsyncExecutor;
import org.apache.aurora.scheduler.async.GatedWorkQueue;
import org.apache.aurora.scheduler.async.GatedWorkQueue.GatedOperation;
+import org.apache.aurora.scheduler.resources.ResourceType;
import org.apache.aurora.scheduler.storage.AttributeStore;
import org.apache.aurora.scheduler.storage.CronJobStore;
import org.apache.aurora.scheduler.storage.JobUpdateStore;
@@ -234,6 +235,10 @@ class DbStorage extends AbstractIdleService implements Storage {
enumValueMapper.addEnumValue("task_states", status.getValue(), status.name());
}
+ for (ResourceType resourceType : ResourceType.values()) {
+ enumValueMapper.addEnumValue("resource_types", resourceType.getValue(), resourceType.name());
+ }
+
createPoolMetrics();
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigManager.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigManager.java b/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigManager.java
index 25160df..17b2490 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigManager.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigManager.java
@@ -20,6 +20,9 @@ import javax.inject.Inject;
import com.google.common.collect.Maps;
+import org.apache.aurora.GuavaUtils;
+import org.apache.aurora.common.collections.Pair;
+import org.apache.aurora.scheduler.resources.ResourceType;
import org.apache.aurora.scheduler.storage.db.views.DbTaskConfig;
import org.apache.aurora.scheduler.storage.db.views.Pairs;
import org.apache.aurora.scheduler.storage.entities.IAppcImage;
@@ -96,6 +99,16 @@ class TaskConfigManager {
}
}
+ if (!config.getResources().isEmpty()) {
+ configMapper.insertResources(
+ configInsert.getId(),
+ config.getResources().stream()
+ .map(e -> Pair.of(
+ ResourceType.fromResource(e).getValue(),
+ ResourceType.fromResource(e).getTypeConverter().stringify(e.getRawValue())))
+ .collect(GuavaUtils.toImmutableList()));
+ }
+
if (!config.getRequestedPorts().isEmpty()) {
configMapper.insertRequestedPorts(configInsert.getId(), config.getRequestedPorts());
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.java b/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.java
index e778a39..04666ad 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.java
@@ -176,4 +176,14 @@ interface TaskConfigMapper extends GarbageCollectedTableMapper {
@Param("configId") long configId,
@Param("name") String name,
@Param("tag") String tag);
+
+ /**
+ * Inserts task resources.
+ *
+ * @param configId Task config ID.
+ * @param values Resources to insert.
+ */
+ void insertResources(
+ @Param("configId") long configId,
+ @Param("values") List<Pair<Integer, String>> values);
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V003_CreateResourceTypesTable.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V003_CreateResourceTypesTable.java b/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V003_CreateResourceTypesTable.java
new file mode 100644
index 0000000..76c6916
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V003_CreateResourceTypesTable.java
@@ -0,0 +1,56 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.db.migration;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import org.apache.aurora.scheduler.resources.ResourceType;
+import org.apache.ibatis.migration.MigrationScript;
+
+public class V003_CreateResourceTypesTable implements MigrationScript {
+ @Override
+ public BigDecimal getId() {
+ return BigDecimal.valueOf(3L);
+ }
+
+ @Override
+ public String getDescription() {
+ return "Create the resource_types table.";
+ }
+ @Override
+ public String getUpScript() {
+ return "CREATE TABLE IF NOT EXISTS resource_types("
+ + "id INT PRIMARY KEY,"
+ + "name VARCHAR NOT NULL,"
+ + "UNIQUE(name)"
+ + ");\n"
+ + populateScript();
+ }
+
+ @Override
+ public String getDownScript() {
+ return "DROP TABLE IF EXISTS resource_types;";
+ }
+
+ private static String populateScript() {
+ return Arrays.stream(ResourceType.values())
+ .map(e -> String.format(
+ "MERGE INTO resource_types(id, name) KEY(name) VALUES (%d, '%s');",
+ e.getValue(),
+ e.name()))
+ .collect(Collectors.joining("\n"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V004_CreateTaskResourceTable.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V004_CreateTaskResourceTable.java b/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V004_CreateTaskResourceTable.java
new file mode 100644
index 0000000..af106a8
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/migration/V004_CreateTaskResourceTable.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.db.migration;
+
+import java.math.BigDecimal;
+
+import org.apache.aurora.scheduler.resources.ResourceType;
+import org.apache.ibatis.migration.MigrationScript;
+
+public class V004_CreateTaskResourceTable implements MigrationScript {
+ @Override
+ public BigDecimal getId() {
+ return BigDecimal.valueOf(4L);
+ }
+
+ @Override
+ public String getDescription() {
+ return "Create the task_resource table.";
+ }
+
+ @Override
+ public String getUpScript() {
+ return "CREATE TABLE IF NOT EXISTS task_resource("
+ + "id IDENTITY,"
+ + "task_config_id BIGINT NOT NULL REFERENCES task_configs(id) ON DELETE CASCADE,"
+ + "type_id INT NOT NULL REFERENCES resource_types(id),"
+ + "value VARCHAR NOT NULL,"
+ + "UNIQUE(task_config_id, type_id, value)"
+ + ");\n"
+ + migrateScript();
+ }
+
+ @Override
+ public String getDownScript() {
+ return "DROP TABLE IF EXISTS task_resource;";
+ }
+
+ private static String migrateScript() {
+ return "MERGE INTO task_resource(task_config_id, type_id, value) "
+ + "KEY(task_config_id, type_id, value) "
+ + "SELECT t.id, rt.id, t.num_cpus FROM task_configs t "
+ + "JOIN resource_types rt ON rt.name = "
+ + String.format("'%s';%n", ResourceType.CPUS.name())
+
+ + "MERGE INTO task_resource(task_config_id, type_id, value) "
+ + "KEY(task_config_id, type_id, value) "
+ + "SELECT t.id, rt.id, t.ram_mb FROM task_configs t "
+ + "JOIN resource_types rt ON rt.NAME = "
+ + String.format("'%s';%n", ResourceType.RAM_MB.name())
+
+ + "MERGE INTO task_resource(task_config_id, type_id, value) "
+ + "KEY(task_config_id, type_id, value) "
+ + "SELECT t.id, rt.id, t.disk_mb FROM task_configs t "
+ + "JOIN resource_types rt ON rt.NAME = "
+ + String.format("'%s';%n", ResourceType.DISK_MB.name())
+
+ + "MERGE INTO task_resource(task_config_id, type_id, value) "
+ + "KEY(task_config_id, type_id, value) "
+ + "SELECT tcrp.task_config_id, rt.id, tcrp.port_name FROM task_config_requested_ports tcrp "
+ + "JOIN resource_types rt ON rt.NAME = "
+ + String.format("'%s';%n", ResourceType.PORTS.name());
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/ResourceTypeHandler.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/ResourceTypeHandler.java b/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/ResourceTypeHandler.java
new file mode 100644
index 0000000..e1bb7fd
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/ResourceTypeHandler.java
@@ -0,0 +1,26 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.db.typehandlers;
+
+import org.apache.aurora.scheduler.resources.ResourceType;
+
+/**
+ * Type handler for {@link ResourceType} enum.
+ */
+public class ResourceTypeHandler extends AbstractTEnumTypeHandler<ResourceType> {
+ @Override
+ protected ResourceType fromValue(int value) {
+ return ResourceType.fromIdValue(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/TypeHandlers.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/TypeHandlers.java b/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/TypeHandlers.java
index ed561c6..e30c387 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/TypeHandlers.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/typehandlers/TypeHandlers.java
@@ -34,6 +34,7 @@ public final class TypeHandlers {
.add(JobUpdateStatusTypeHandler.class)
.add(MaintenanceModeTypeHandler.class)
.add(ScheduleStatusTypeHandler.class)
+ .add(ResourceTypeHandler.class)
.build();
}
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/views/DBResource.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/views/DBResource.java b/src/main/java/org/apache/aurora/scheduler/storage/db/views/DBResource.java
new file mode 100644
index 0000000..dc7e97d
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/views/DBResource.java
@@ -0,0 +1,30 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.db.views;
+
+import org.apache.aurora.gen.Resource;
+import org.apache.aurora.scheduler.resources.ResourceType;
+import org.apache.aurora.scheduler.storage.entities.IResource;
+
+public final class DBResource {
+ private ResourceType type;
+ private String value;
+
+ private DBResource() {
+ }
+
+ Resource toThrift() {
+ return IResource.newBuilder(type.getValue(), type.getTypeConverter().parseFrom(value));
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/db/views/DbTaskConfig.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/db/views/DbTaskConfig.java b/src/main/java/org/apache/aurora/scheduler/storage/db/views/DbTaskConfig.java
index cdd1060..a7523c4 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/db/views/DbTaskConfig.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/db/views/DbTaskConfig.java
@@ -17,7 +17,6 @@ import java.util.List;
import com.google.common.collect.ImmutableSet;
-import org.apache.aurora.GuavaUtils;
import org.apache.aurora.common.collections.Pair;
import org.apache.aurora.gen.Container;
import org.apache.aurora.gen.ExecutorConfig;
@@ -28,6 +27,8 @@ import org.apache.aurora.gen.Metadata;
import org.apache.aurora.gen.TaskConfig;
import org.apache.aurora.scheduler.storage.entities.ITaskConfig;
+import static org.apache.aurora.GuavaUtils.toImmutableSet;
+
public final class DbTaskConfig {
private long rowId;
private JobKey job;
@@ -48,6 +49,7 @@ public final class DbTaskConfig {
private DbContainer container;
private String tier;
private DbImage image;
+ private List<DBResource> resources;
private DbTaskConfig() {
}
@@ -71,14 +73,15 @@ public final class DbTaskConfig {
.setImage(image == null ? null : image.toThrift())
.setConstraints(constraints.stream()
.map(DbConstraint::toThrift)
- .collect(GuavaUtils.toImmutableSet()))
+ .collect(toImmutableSet()))
.setRequestedPorts(ImmutableSet.copyOf(requestedPorts))
.setTaskLinks(Pairs.toMap(taskLinks))
.setContactEmail(contactEmail)
.setExecutorConfig(executorConfig)
.setMetadata(ImmutableSet.copyOf(metadata))
.setContainer(
- container == null ? Container.mesos(new MesosContainer()) : container.toThrift());
+ container == null ? Container.mesos(new MesosContainer()) : container.toThrift())
+ .setResources(resources.stream().map(DBResource::toThrift).collect(toImmutableSet()));
}
public ITaskConfig toImmutable() {
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/log/LogStorage.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/log/LogStorage.java b/src/main/java/org/apache/aurora/scheduler/storage/log/LogStorage.java
index f586186..39b6567 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/log/LogStorage.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/log/LogStorage.java
@@ -35,7 +35,6 @@ import org.apache.aurora.common.quantity.Amount;
import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.common.stats.SlidingStats;
import org.apache.aurora.gen.HostAttributes;
-import org.apache.aurora.gen.JobUpdate;
import org.apache.aurora.gen.storage.LogEntry;
import org.apache.aurora.gen.storage.Op;
import org.apache.aurora.gen.storage.RewriteTask;
@@ -62,16 +61,13 @@ import org.apache.aurora.scheduler.storage.Storage.MutateWork.NoResult;
import org.apache.aurora.scheduler.storage.Storage.NonVolatileStorage;
import org.apache.aurora.scheduler.storage.TaskStore;
import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
-import org.apache.aurora.scheduler.storage.entities.IJobConfiguration;
import org.apache.aurora.scheduler.storage.entities.IJobInstanceUpdateEvent;
import org.apache.aurora.scheduler.storage.entities.IJobKey;
-import org.apache.aurora.scheduler.storage.entities.IJobUpdate;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateEvent;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateKey;
import org.apache.aurora.scheduler.storage.entities.ILock;
import org.apache.aurora.scheduler.storage.entities.ILockKey;
import org.apache.aurora.scheduler.storage.entities.IResourceAggregate;
-import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
import org.apache.aurora.scheduler.storage.entities.ITaskConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -330,7 +326,7 @@ public class LogStorage implements NonVolatileStorage, DistributedSnapshotStore
.put(Op._Fields.SAVE_CRON_JOB, op -> {
SaveCronJob cronJob = op.getSaveCronJob();
writeBehindJobStore.saveAcceptedJob(
- IJobConfiguration.build(cronJob.getJobConfig()));
+ ThriftBackfill.backfillJobConfiguration(cronJob.getJobConfig()));
})
.put(
Op._Fields.REMOVE_JOB,
@@ -338,7 +334,7 @@ public class LogStorage implements NonVolatileStorage, DistributedSnapshotStore
.put(
Op._Fields.SAVE_TASKS,
op -> writeBehindTaskStore.saveTasks(
- IScheduledTask.setFromBuilders(op.getSaveTasks().getTasks())))
+ ThriftBackfill.backfillTasks(op.getSaveTasks().getTasks())))
.put(Op._Fields.REWRITE_TASK, op -> {
RewriteTask rewriteTask = op.getRewriteTask();
writeBehindTaskStore.unsafeModifyInPlace(
@@ -374,12 +370,10 @@ public class LogStorage implements NonVolatileStorage, DistributedSnapshotStore
.put(
Op._Fields.REMOVE_LOCK,
op -> writeBehindLockStore.removeLock(ILockKey.build(op.getRemoveLock().getLockKey())))
- .put(Op._Fields.SAVE_JOB_UPDATE, op -> {
- JobUpdate update = op.getSaveJobUpdate().getJobUpdate();
+ .put(Op._Fields.SAVE_JOB_UPDATE, op ->
writeBehindJobUpdateStore.saveJobUpdate(
- IJobUpdate.build(update),
- Optional.fromNullable(op.getSaveJobUpdate().getLockToken()));
- })
+ ThriftBackfill.backFillJobUpdate(op.getSaveJobUpdate().getJobUpdate()),
+ Optional.fromNullable(op.getSaveJobUpdate().getLockToken())))
.put(Op._Fields.SAVE_JOB_UPDATE_EVENT, op -> {
SaveJobUpdateEvent event = op.getSaveJobUpdateEvent();
writeBehindJobUpdateStore.saveJobUpdateEvent(
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImpl.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImpl.java b/src/main/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImpl.java
index b6922e1..8eed1fc 100644
--- a/src/main/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImpl.java
+++ b/src/main/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImpl.java
@@ -60,7 +60,6 @@ import org.apache.aurora.scheduler.storage.db.MigrationManager;
import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
import org.apache.aurora.scheduler.storage.entities.IJobConfiguration;
import org.apache.aurora.scheduler.storage.entities.IJobInstanceUpdateEvent;
-import org.apache.aurora.scheduler.storage.entities.IJobUpdate;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateEvent;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateKey;
import org.apache.aurora.scheduler.storage.entities.ILock;
@@ -222,8 +221,8 @@ public class SnapshotStoreImpl implements SnapshotStore<Snapshot> {
store.getUnsafeTaskStore().deleteAllTasks();
if (snapshot.isSetTasks()) {
- store.getUnsafeTaskStore().saveTasks(
- IScheduledTask.setFromBuilders(snapshot.getTasks()));
+ store.getUnsafeTaskStore()
+ .saveTasks(ThriftBackfill.backfillTasks(snapshot.getTasks()));
}
}
},
@@ -251,7 +250,7 @@ public class SnapshotStoreImpl implements SnapshotStore<Snapshot> {
if (snapshot.isSetCronJobs()) {
for (StoredCronJob job : snapshot.getCronJobs()) {
store.getCronJobStore().saveAcceptedJob(
- IJobConfiguration.build(job.getJobConfiguration()));
+ ThriftBackfill.backfillJobConfiguration(job.getJobConfiguration()));
}
}
}
@@ -328,7 +327,7 @@ public class SnapshotStoreImpl implements SnapshotStore<Snapshot> {
for (StoredJobUpdateDetails storedDetails : snapshot.getJobUpdateDetails()) {
JobUpdateDetails details = storedDetails.getDetails();
updateStore.saveJobUpdate(
- IJobUpdate.build(details.getUpdate()),
+ ThriftBackfill.backFillJobUpdate(details.getUpdate()),
Optional.fromNullable(storedDetails.getLockToken()));
if (details.getUpdateEventsSize() > 0) {
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/java/org/apache/aurora/scheduler/storage/log/ThriftBackfill.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/storage/log/ThriftBackfill.java b/src/main/java/org/apache/aurora/scheduler/storage/log/ThriftBackfill.java
new file mode 100644
index 0000000..9c86aa0
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/storage/log/ThriftBackfill.java
@@ -0,0 +1,127 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.log;
+
+import java.util.Set;
+
+import org.apache.aurora.GuavaUtils;
+import org.apache.aurora.gen.JobConfiguration;
+import org.apache.aurora.gen.JobUpdate;
+import org.apache.aurora.gen.JobUpdateInstructions;
+import org.apache.aurora.gen.Resource;
+import org.apache.aurora.gen.ScheduledTask;
+import org.apache.aurora.gen.TaskConfig;
+import org.apache.aurora.scheduler.resources.ResourceType;
+import org.apache.aurora.scheduler.storage.entities.IJobConfiguration;
+import org.apache.aurora.scheduler.storage.entities.IJobUpdate;
+import org.apache.aurora.scheduler.storage.entities.IResource;
+import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
+
+import static org.apache.aurora.scheduler.resources.ResourceType.CPUS;
+import static org.apache.aurora.scheduler.resources.ResourceType.DISK_MB;
+import static org.apache.aurora.scheduler.resources.ResourceType.PORTS;
+import static org.apache.aurora.scheduler.resources.ResourceType.RAM_MB;
+
+/**
+ * Helps migrating thrift schema by populating deprecated and/or replacement fields.
+ */
+public final class ThriftBackfill {
+ private ThriftBackfill() {
+ // Utility class.
+ }
+
+ private static Resource getResource(TaskConfig config, ResourceType type) {
+ return config.getResources().stream()
+ .filter(e -> ResourceType.fromResource(IResource.build(e)).equals(type))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Missing resource definition for " + type));
+ }
+
+ /**
+ * Ensures TaskConfig.resources and correspondent task-level fields are all populated.
+ *
+ * @param config TaskConfig to backfill.
+ * @return Backfilled TaskConfig.
+ */
+ public static TaskConfig backfillTask(TaskConfig config) {
+ if (!config.isSetResources() || config.getResources().isEmpty()) {
+ config.addToResources(Resource.numCpus(config.getNumCpus()));
+ config.addToResources(Resource.ramMb(config.getRamMb()));
+ config.addToResources(Resource.diskMb(config.getDiskMb()));
+ if (config.isSetRequestedPorts()) {
+ for (String port : config.getRequestedPorts()) {
+ config.addToResources(Resource.namedPort(port));
+ }
+ }
+ } else {
+ config.setNumCpus(getResource(config, CPUS).getNumCpus());
+ config.setRamMb(getResource(config, RAM_MB).getRamMb());
+ config.setDiskMb(getResource(config, DISK_MB).getDiskMb());
+ Set<String> ports = config.getResources().stream()
+ .filter(e -> ResourceType.fromResource(IResource.build(e)).equals(PORTS))
+ .map(Resource::getNamedPort)
+ .collect(GuavaUtils.toImmutableSet());
+ if (!ports.isEmpty()) {
+ config.setRequestedPorts(ports);
+ }
+ }
+ return config;
+ }
+
+ /**
+ * Backfills JobConfiguration. See {@link #backfillTask(TaskConfig)}.
+ *
+ * @param jobConfig JobConfiguration to backfill.
+ * @return Backfilled JobConfiguration.
+ */
+ public static IJobConfiguration backfillJobConfiguration(JobConfiguration jobConfig) {
+ backfillTask(jobConfig.getTaskConfig());
+ return IJobConfiguration.build(jobConfig);
+ }
+
+ /**
+ * Backfills set of tasks. See {@link #backfillTask(TaskConfig)}.
+ *
+ * @param tasks Set of tasks to backfill.
+ * @return Backfilled set of tasks.
+ */
+ public static Set<IScheduledTask> backfillTasks(Set<ScheduledTask> tasks) {
+ return tasks.stream()
+ .map(ThriftBackfill::backfillScheduledTask)
+ .map(IScheduledTask::build)
+ .collect(GuavaUtils.toImmutableSet());
+ }
+
+ private static ScheduledTask backfillScheduledTask(ScheduledTask task) {
+ backfillTask(task.getAssignedTask().getTask());
+ return task;
+ }
+
+ /**
+ * Backfills JobUpdate. See {@link #backfillTask(TaskConfig)}.
+ *
+ * @param update JobUpdate to backfill.
+ * @return Backfilled job update.
+ */
+ static IJobUpdate backFillJobUpdate(JobUpdate update) {
+ JobUpdateInstructions instructions = update.getInstructions();
+ if (instructions.isSetDesiredState()) {
+ backfillTask(instructions.getDesiredState().getTask());
+ }
+
+ instructions.getInitialState().forEach(e -> backfillTask(e.getTask()));
+
+ return IJobUpdate.build(update);
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/python/apache/aurora/config/thrift.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/config/thrift.py b/src/main/python/apache/aurora/config/thrift.py
index be0cd68..928ca93 100644
--- a/src/main/python/apache/aurora/config/thrift.py
+++ b/src/main/python/apache/aurora/config/thrift.py
@@ -35,6 +35,7 @@ from gen.apache.aurora.api.ttypes import (
LimitConstraint,
MesosContainer,
Metadata,
+ Resource,
TaskConfig,
TaskConstraint,
ValueConstraint
@@ -218,6 +219,10 @@ def convert(job, metadata=frozenset(), ports=frozenset()):
raise InvalidConfig('Task has invalid resources. cpu/ramMb/diskMb must all be positive: '
'cpu:%r ramMb:%r diskMb:%r' % (task.numCpus, task.ramMb, task.diskMb))
+ task.resources = frozenset(
+ [Resource(numCpus=task.numCpus), Resource(ramMb=task.ramMb), Resource(diskMb=task.diskMb)] +
+ [Resource(namedPort=p) for p in ports])
+
task.job = key
task.owner = owner
task.requestedPorts = ports
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/python/apache/aurora/tools/java/thrift_wrapper_codegen.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/tools/java/thrift_wrapper_codegen.py b/src/main/python/apache/aurora/tools/java/thrift_wrapper_codegen.py
index 3465fe9..7c28180 100644
--- a/src/main/python/apache/aurora/tools/java/thrift_wrapper_codegen.py
+++ b/src/main/python/apache/aurora/tools/java/thrift_wrapper_codegen.py
@@ -127,12 +127,21 @@ UNION_FIELD_TEMPLATE = ''' public %(type)s %(fn_name)s() {
UNION_SWITCH_CASE = '''case %(case)s:
%(body)s'''
-UNION_FIELD_SWITCH = '''switch (getSetField()) {
+UNION_DEFAULT_ERROR = 'throw new RuntimeException("Unrecognized field " + getSetField())';
+UNION_FIELD_SWITCH = '''switch (%(switch_by)s) {
%(cases)s
default:
- throw new RuntimeException("Unrecognized field " + getSetField());
+ %(error)s;
}'''
+UNION_COPY_CONSTRUCTOR_2 = ''' public static %(wrapped)s newBuilder(int id, Object value) {
+ %(body)s
+ }'''
+
+UNION_VALUE_ACCESSOR = ''' public Object getRawValue() {
+ return value;
+ }'''
+
SIMPLE_ASSIGNMENT = 'this.%(field)s = wrapped.%(fn_name)s();'
FIELD_DECLARATION = '''private final %(type)s %(field)s;'''
@@ -142,7 +151,7 @@ STRUCT_ASSIGNMENT = '''this.%(field)s = wrapped.%(isset)s()
IMMUTABLE_COLLECTION_DECLARATION = (
- '''private final Immutable%(collection)s<%(params)s> %(field)s;''')
+ '''private final Immutable%(collection)s<%(params)s> %(field)s;''')
IMMUTABLE_COLLECTION_ASSIGNMENT = '''this.%(field)s = wrapped.%(isset)s()
? Immutable%(collection)s.copyOf(wrapped.%(fn_name)s())
: Immutable%(collection)s.of();'''
@@ -481,7 +490,7 @@ def generate_union_field(code, struct, field):
'enum_value': field_enum_value})
-def generate_struct_field(code, struct, field, builder_calls):
+def generate_struct_field(code, field, builder_calls):
field_type = field.ttype.codegen_name()
assignment = SIMPLE_ASSIGNMENT
assignment_args = {
@@ -574,28 +583,58 @@ def generate_java(struct):
if struct.kind == 'union':
assign_cases = []
copy_cases = []
+ copy_2_cases = []
for field in struct.fields:
generate_union_field(code, struct, field)
- assign_case_body = 'value = %(codegen_name)s.build(wrapped.%(accessor_method)s());\nbreak;' % {
+ assert field.ttype.immutable or isinstance(field.ttype, StructType), 'Unrecognized type %s' % field.ttype.name
+
+ if field.ttype.immutable:
+ assign_case_body = 'value = wrapped.%s();\nbreak;' % field.accessor_method()
+ copy_case_body = 'return new %s(setField, %s());' % (struct.name, field.accessor_method())
+ copy_2_case_cast = field.ttype.codegen_name()
+ else:
+ assign_case_body = 'value = %(codegen_name)s.build(wrapped.%(accessor_method)s());\nbreak;' % {
'codegen_name': field.ttype.codegen_name(),
'accessor_method': field.accessor_method()}
+ copy_case_body = 'return new %s(setField, %s().newBuilder());' % (struct.name, field.accessor_method())
+ code.add_import('org.apache.aurora.gen.%s' % field.ttype.name)
+ copy_2_case_cast = field.ttype.name
+
assign_cases.append(UNION_SWITCH_CASE % {'case': to_upper_snake_case(field.name),
'body': assign_case_body})
- copy_case_body = 'return new %s(setField, %s().newBuilder());' % (struct.name, field.accessor_method())
copy_cases.append(UNION_SWITCH_CASE % {'case': to_upper_snake_case(field.name),
'body': copy_case_body})
+ copy_2_case_body = 'return %(wrapped)s.%(method)s((%(cast)s) value);' % {
+ 'wrapped': struct.name,
+ 'method': field.name,
+ 'cast': copy_2_case_cast}
+
+ copy_2_cases.append(UNION_SWITCH_CASE % {'case': to_upper_snake_case(field.name),
+ 'body': copy_2_case_body})
+
set_field_type = '%s._Fields' % struct.name
code.add_accessor(FIELD_TEMPLATE % {'type': set_field_type, 'fn_name': 'getSetField', 'field': 'setField'})
code.add_field(FIELD_DECLARATION % {'field': 'setField', 'type': set_field_type})
code.add_assignment(SIMPLE_ASSIGNMENT % {'field': 'setField',
'fn_name': 'getSetField'})
code.add_field(FIELD_DECLARATION % {'field': 'value', 'type': 'Object'})
- code.add_assignment(UNION_FIELD_SWITCH % {'cases': '\n '.join(assign_cases)})
+ code.add_assignment(UNION_FIELD_SWITCH % {'cases': '\n '.join(assign_cases),
+ 'switch_by': 'getSetField()',
+ 'error': UNION_DEFAULT_ERROR})
+
+ code.copy_constructor = UNION_FIELD_SWITCH % {'cases': '\n '.join(copy_cases),
+ 'switch_by': 'getSetField()',
+ 'error': UNION_DEFAULT_ERROR}
+
+ copy_2_switch = UNION_FIELD_SWITCH % {'cases': '\n '.join(copy_2_cases),
+ 'switch_by': '%s.findByThriftId(id)' % set_field_type,
+ 'error': 'throw new RuntimeException("Unrecognized id " + id)'}
- code.copy_constructor = UNION_FIELD_SWITCH % {'cases': '\n '.join(copy_cases)}
+ code.add_accessor(UNION_VALUE_ACCESSOR)
+ code.add_accessor(UNION_COPY_CONSTRUCTOR_2 % {'wrapped': struct.name, 'body': copy_2_switch})
code.to_string = '.add("setField", setField).add("value", value)'
code.equals = 'Objects.equals(setField, other.setField) && Objects.equals(value, other.value)'
@@ -603,7 +642,7 @@ def generate_java(struct):
else:
builder_calls = []
for field in struct.fields:
- generate_struct_field(code, struct, field, builder_calls)
+ generate_struct_field(code, field, builder_calls)
field_names = [f.name for f in struct.fields]
code.copy_constructor = 'return new %s()%s;' % (struct.name, '\n ' + '\n '.join(builder_calls))
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/resources/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.xml
----------------------------------------------------------------------
diff --git a/src/main/resources/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.xml b/src/main/resources/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.xml
index 36df7ed..4162797 100644
--- a/src/main/resources/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.xml
+++ b/src/main/resources/org/apache/aurora/scheduler/storage/db/TaskConfigMapper.xml
@@ -125,6 +125,13 @@
<id column="id" />
</resultMap>
+ <resultMap id="resourceMap" type="org.apache.aurora.scheduler.storage.db.views.DBResource">
+ <id column="id" />
+ <result property="type"
+ column="type_id"
+ typeHandler="org.apache.aurora.scheduler.storage.db.typehandlers.ResourceTypeHandler" />
+ </resultMap>
+
<resultMap id="taskConfigMap" type="org.apache.aurora.scheduler.storage.db.views.DbTaskConfig">
<id column="id" property="rowId" />
<result column="creator_user" property="owner.user"/>
@@ -149,6 +156,12 @@
select="selectTaskLinks"
column="id"
foreignColumn="task_config_id"/>
+ <!-- TODO(maxim): move resources to a main join when task level fields are removed. -->
+ <collection
+ property="resources"
+ select="selectResources"
+ column="id"
+ foreignColumn="task_config_id" />
</resultMap>
<sql id="unscopedConfigSelect">
@@ -265,6 +278,20 @@
)
</insert>
+ <insert id="insertResources">
+ INSERT INTO task_resource (
+ task_config_id,
+ type_id,
+ value
+ ) VALUES (
+ <foreach item="value" collection="values" separator="),(">
+ #{configId},
+ #{value.first},
+ #{value.second}
+ </foreach>
+ )
+ </insert>
+
<insert id="insertTaskLinks" >
INSERT INTO task_config_task_links (
task_config_id,
@@ -295,6 +322,15 @@
WHERE task_config_id = #{id}
</select>
+ <select id="selectResources" resultMap="resourceMap">
+ SELECT
+ id,
+ type_id,
+ value
+ FROM task_resource
+ WHERE task_config_id = #{id}
+ </select>
+
<insert id="insertContainer" useGeneratedKeys="true" keyColumn="id" keyProperty="result.id">
INSERT INTO task_config_docker_containers (
task_config_id,
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/main/resources/org/apache/aurora/scheduler/storage/db/schema.sql
----------------------------------------------------------------------
diff --git a/src/main/resources/org/apache/aurora/scheduler/storage/db/schema.sql b/src/main/resources/org/apache/aurora/scheduler/storage/db/schema.sql
index 92a0798..45ec1bf 100644
--- a/src/main/resources/org/apache/aurora/scheduler/storage/db/schema.sql
+++ b/src/main/resources/org/apache/aurora/scheduler/storage/db/schema.sql
@@ -100,6 +100,22 @@ CREATE TABLE task_configs(
tier VARCHAR
);
+CREATE TABLE resource_types(
+ id INT PRIMARY KEY,
+ name VARCHAR NOT NULL,
+
+ UNIQUE(name)
+);
+
+CREATE TABLE task_resource(
+ id IDENTITY,
+ task_config_id BIGINT NOT NULL REFERENCES task_configs(id) ON DELETE CASCADE,
+ type_id INT NOT NULL REFERENCES resource_types(id),
+ value VARCHAR NOT NULL,
+
+ UNIQUE(task_config_id, type_id, value)
+);
+
CREATE TABLE task_constraints(
id IDENTITY,
task_config_id BIGINT NOT NULL REFERENCES task_configs(id) ON DELETE CASCADE,
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
index c310690..1e9e1ae 100644
--- a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
@@ -48,6 +48,8 @@ import org.apache.aurora.common.zookeeper.Credentials;
import org.apache.aurora.common.zookeeper.ServerSetImpl;
import org.apache.aurora.common.zookeeper.ZooKeeperClient;
import org.apache.aurora.common.zookeeper.testing.BaseZooKeeperClientTest;
+import org.apache.aurora.gen.HostAttributes;
+import org.apache.aurora.gen.MaintenanceMode;
import org.apache.aurora.gen.ScheduleStatus;
import org.apache.aurora.gen.ScheduledTask;
import org.apache.aurora.gen.ServerInfo;
@@ -73,6 +75,7 @@ import org.apache.aurora.scheduler.mesos.DriverSettings;
import org.apache.aurora.scheduler.mesos.TestExecutorSettings;
import org.apache.aurora.scheduler.resources.ResourceSlot;
import org.apache.aurora.scheduler.storage.backup.BackupModule;
+import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
import org.apache.aurora.scheduler.storage.entities.IServerInfo;
import org.apache.aurora.scheduler.storage.log.EntrySerializer;
@@ -113,6 +116,11 @@ public class SchedulerIT extends BaseZooKeeperClientTest {
private static final String SERVERSET_PATH = "/fake/service/path";
private static final String STATS_URL_PREFIX = "fake_url";
private static final String FRAMEWORK_ID = "integration_test_framework_id";
+ private static final IHostAttributes HOST_ATTRIBUTES = IHostAttributes.build(new HostAttributes()
+ .setHost("host")
+ .setSlaveId("slave-id")
+ .setMode(MaintenanceMode.NONE)
+ .setAttributes(ImmutableSet.of()));
private static final DriverSettings SETTINGS = new DriverSettings(
"fakemaster",
@@ -276,7 +284,9 @@ public class SchedulerIT extends BaseZooKeeperClientTest {
status,
100)
.newBuilder();
- builder.getAssignedTask().setSlaveId("slave-id");
+ builder.getAssignedTask()
+ .setSlaveId(HOST_ATTRIBUTES.getSlaveId())
+ .setSlaveHost(HOST_ATTRIBUTES.getHost());
return IScheduledTask.build(builder);
}
@@ -293,7 +303,9 @@ public class SchedulerIT extends BaseZooKeeperClientTest {
IScheduledTask snapshotTask = makeTask("snapshotTask", ScheduleStatus.ASSIGNED);
IScheduledTask transactionTask = makeTask("transactionTask", ScheduleStatus.RUNNING);
Iterable<Entry> recoveredEntries = toEntries(
- LogEntry.snapshot(new Snapshot().setTasks(ImmutableSet.of(snapshotTask.newBuilder()))),
+ LogEntry.snapshot(new Snapshot()
+ .setTasks(ImmutableSet.of(snapshotTask.newBuilder()))
+ .setHostAttributes(ImmutableSet.of(HOST_ATTRIBUTES.newBuilder()))),
LogEntry.transaction(new Transaction(
ImmutableList.of(Op.saveTasks(
new SaveTasks(ImmutableSet.of(transactionTask.newBuilder())))),
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java b/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
index 1ccfe01..7313279 100644
--- a/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
@@ -44,11 +44,16 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import static org.apache.aurora.gen.Resource.diskMb;
+import static org.apache.aurora.gen.Resource.namedPort;
+import static org.apache.aurora.gen.Resource.numCpus;
+import static org.apache.aurora.gen.Resource.ramMb;
import static org.apache.aurora.gen.test.testConstants.INVALID_IDENTIFIERS;
import static org.apache.aurora.gen.test.testConstants.VALID_IDENTIFIERS;
import static org.apache.aurora.scheduler.base.UserProvidedStrings.isGoodIdentifier;
import static org.apache.aurora.scheduler.configuration.ConfigurationManager.DEDICATED_ATTRIBUTE;
import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -100,7 +105,11 @@ public class ConfigurationManagerTest {
.setName(DEDICATED_ATTRIBUTE)
.setConstraint(TaskConstraint.value(new ValueConstraint(
false, ImmutableSet.of("foo"))))))
- .setOwner(new Identity().setUser("owner-user")));
+ .setOwner(new Identity().setUser("owner-user"))
+ .setResources(ImmutableSet.of(
+ numCpus(1.0),
+ ramMb(1),
+ diskMb(1))));
private static final ITaskConfig CONFIG_WITH_CONTAINER =
TaskTestUtil.makeConfig(JobKeys.from("role", "env", "job"));
@@ -260,6 +269,35 @@ public class ConfigurationManagerTest {
CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
}
+ @Test
+ public void testTaskResourceBackfill() throws Exception {
+ TaskConfig builder = CONFIG_WITH_CONTAINER.newBuilder();
+ builder.unsetResources();
+
+ assertFalse(builder.isSetResources());
+ ITaskConfig populated =
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ assertEquals(CONFIG_WITH_CONTAINER.getResources(), populated.getResources());
+ }
+
+ @Test
+ public void testMultipleResourceValuesBlocked() throws Exception {
+ TaskConfig builder = CONFIG_WITH_CONTAINER.newBuilder();
+ builder.addToResources(numCpus(3.0));
+ builder.addToResources(ramMb(72));
+
+ expectTaskDescriptionException("Multiple resource values are not supported for CPU, RAM");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ }
+
+ @Test
+ public void testMultipleResourceValuesAllowed() throws Exception {
+ TaskConfig builder = CONFIG_WITH_CONTAINER.newBuilder();
+ builder.addToResources(namedPort("thrift"));
+
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ }
+
private void expectTaskDescriptionException(String message) {
expectedException.expect(TaskDescriptionException.class);
expectedException.expectMessage(message);
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeConverterTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeConverterTest.java b/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeConverterTest.java
new file mode 100644
index 0000000..78da84a
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeConverterTest.java
@@ -0,0 +1,30 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.resources;
+
+import org.junit.Test;
+
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.DOUBLE;
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.LONG;
+import static org.apache.aurora.scheduler.resources.ResourceTypeConverter.STRING;
+import static org.junit.Assert.assertEquals;
+
+public class ResourceTypeConverterTest {
+ @Test
+ public void testRoundtrip() {
+ assertEquals(234L, LONG.parseFrom(LONG.stringify(234L)).longValue());
+ assertEquals(2.34, DOUBLE.parseFrom(DOUBLE.stringify(2.34)).doubleValue(), 0.0);
+ assertEquals("http", STRING.parseFrom(STRING.stringify("http")));
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeTest.java b/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeTest.java
new file mode 100644
index 0000000..dc9dc66
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/resources/ResourceTypeTest.java
@@ -0,0 +1,36 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.resources;
+
+import org.apache.aurora.gen.Resource;
+import org.apache.aurora.scheduler.storage.entities.IResource;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class ResourceTypeTest {
+ @Test
+ public void testFindValueById() {
+ assertEquals(
+ ResourceType.CPUS,
+ ResourceType.fromIdValue(Resource.numCpus(1.0).getSetField().getThriftFieldId()));
+ }
+
+ @Test
+ public void testFindByResource() {
+ assertEquals(
+ ResourceType.CPUS,
+ ResourceType.fromResource(IResource.build(Resource.numCpus(1.0))));
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/storage/backup/RecoveryTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/storage/backup/RecoveryTest.java b/src/test/java/org/apache/aurora/scheduler/storage/backup/RecoveryTest.java
index a33f6f7..e870087 100644
--- a/src/test/java/org/apache/aurora/scheduler/storage/backup/RecoveryTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/storage/backup/RecoveryTest.java
@@ -112,7 +112,7 @@ public class RecoveryTest extends EasyMockTest {
recovery.stage(backup1);
assertEquals(
IScheduledTask.setFromBuilders(SNAPSHOT1.getTasks()),
- recovery.query(Query.unscoped()));
+ ImmutableSet.copyOf(recovery.query(Query.unscoped())));
recovery.commit();
transaction.getValue().apply(storeProvider);
@@ -138,11 +138,11 @@ public class RecoveryTest extends EasyMockTest {
recovery.stage(backup1);
assertEquals(
IScheduledTask.setFromBuilders(SNAPSHOT1.getTasks()),
- recovery.query(Query.unscoped()));
+ ImmutableSet.copyOf(recovery.query(Query.unscoped())));
recovery.deleteTasks(Query.taskScoped(Tasks.id(TASK2)));
assertEquals(
IScheduledTask.setFromBuilders(modified.getTasks()),
- recovery.query(Query.unscoped()));
+ ImmutableSet.copyOf(recovery.query(Query.unscoped())));
recovery.commit();
transaction.getValue().apply(storeProvider);
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/storage/log/LogStorageTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/storage/log/LogStorageTest.java b/src/test/java/org/apache/aurora/scheduler/storage/log/LogStorageTest.java
index bf9479d..7ac0b58 100644
--- a/src/test/java/org/apache/aurora/scheduler/storage/log/LogStorageTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/storage/log/LogStorageTest.java
@@ -113,6 +113,8 @@ import org.easymock.Capture;
import org.junit.Before;
import org.junit.Test;
+import static org.apache.aurora.scheduler.base.TaskTestUtil.makeConfig;
+import static org.apache.aurora.scheduler.base.TaskTestUtil.makeTask;
import static org.easymock.EasyMock.capture;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
@@ -224,8 +226,7 @@ public class LogStorageTest extends EasyMockTest {
schedulingService.doEvery(eq(SNAPSHOT_INTERVAL), capture(snapshotAction));
Snapshot snapshotContents = new Snapshot()
.setTimestamp(NOW)
- .setTasks(ImmutableSet.of(
- TaskTestUtil.makeTask("task_id", TaskTestUtil.JOB).newBuilder()));
+ .setTasks(ImmutableSet.of(makeTask("task_id", TaskTestUtil.JOB).newBuilder()));
expect(snapshotStore.createSnapshot()).andReturn(snapshotContents);
DeduplicatedSnapshot deduplicated =
new SnapshotDeduplicatorImpl().deduplicate(snapshotContents);
@@ -273,17 +274,23 @@ public class LogStorageTest extends EasyMockTest {
builder.add(createTransaction(Op.saveFrameworkId(new SaveFrameworkId("bob"))));
storageUtil.schedulerStore.saveFrameworkId("bob");
- SaveCronJob cronJob = new SaveCronJob().setJobConfig(new JobConfiguration());
+ JobConfiguration actualJob = new JobConfiguration().setTaskConfig(nonBackfilledConfig());
+ JobConfiguration expectedJob =
+ new JobConfiguration().setTaskConfig(makeConfig(JOB_KEY).newBuilder());
+ SaveCronJob cronJob = new SaveCronJob().setJobConfig(actualJob);
builder.add(createTransaction(Op.saveCronJob(cronJob)));
- storageUtil.jobStore.saveAcceptedJob(IJobConfiguration.build(cronJob.getJobConfig()));
+ storageUtil.jobStore.saveAcceptedJob(IJobConfiguration.build(expectedJob));
RemoveJob removeJob = new RemoveJob(JOB_KEY.newBuilder());
builder.add(createTransaction(Op.removeJob(removeJob)));
storageUtil.jobStore.removeJob(JOB_KEY);
- SaveTasks saveTasks = new SaveTasks(ImmutableSet.of(new ScheduledTask()));
+ ScheduledTask actualTask = makeTask("id", JOB_KEY).newBuilder();
+ actualTask.getAssignedTask().setTask(nonBackfilledConfig());
+ IScheduledTask expectedTask = makeTask("id", JOB_KEY);
+ SaveTasks saveTasks = new SaveTasks(ImmutableSet.of(actualTask));
builder.add(createTransaction(Op.saveTasks(saveTasks)));
- storageUtil.taskStore.saveTasks(IScheduledTask.setFromBuilders(saveTasks.getTasks()));
+ storageUtil.taskStore.saveTasks(ImmutableSet.of(expectedTask));
RewriteTask rewriteTask = new RewriteTask("id1", new TaskConfig());
builder.add(createTransaction(Op.rewriteTask(rewriteTask)));
@@ -326,12 +333,20 @@ public class LogStorageTest extends EasyMockTest {
builder.add(createTransaction(Op.removeLock(removeLock)));
storageUtil.lockStore.removeLock(ILockKey.build(removeLock.getLockKey()));
- JobUpdate update = new JobUpdate().setSummary(
- new JobUpdateSummary().setKey(UPDATE_ID.newBuilder()));
- SaveJobUpdate saveUpdate = new SaveJobUpdate(update, "token");
+ JobUpdate actualUpdate = new JobUpdate()
+ .setSummary(new JobUpdateSummary().setKey(UPDATE_ID.newBuilder()))
+ .setInstructions(new JobUpdateInstructions()
+ .setInitialState(
+ ImmutableSet.of(new InstanceTaskConfig().setTask(nonBackfilledConfig())))
+ .setDesiredState(new InstanceTaskConfig().setTask(nonBackfilledConfig())));
+ JobUpdate expectedUpdate = actualUpdate.deepCopy();
+ expectedUpdate.getInstructions().getDesiredState().setTask(makeConfig(JOB_KEY).newBuilder());
+ expectedUpdate.getInstructions().getInitialState()
+ .forEach(e -> e.setTask(makeConfig(JOB_KEY).newBuilder()));
+ SaveJobUpdate saveUpdate = new SaveJobUpdate(actualUpdate, "token");
builder.add(createTransaction(Op.saveJobUpdate(saveUpdate)));
storageUtil.jobUpdateStore.saveJobUpdate(
- IJobUpdate.build(saveUpdate.getJobUpdate()),
+ IJobUpdate.build(expectedUpdate),
Optional.of(saveUpdate.getLockToken()));
SaveJobUpdateEvent saveUpdateEvent =
@@ -371,6 +386,12 @@ public class LogStorageTest extends EasyMockTest {
expect(stream.readAll()).andReturn(entryBuilder.build().iterator());
}
+ private TaskConfig nonBackfilledConfig() {
+ TaskConfig config = makeConfig(JOB_KEY).newBuilder();
+ config.unsetResources();
+ return config;
+ }
+
abstract class AbstractStorageFixture {
private final AtomicBoolean runCalled = new AtomicBoolean(false);
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImplIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImplIT.java b/src/test/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImplIT.java
index ff9c1d0..05168ba 100644
--- a/src/test/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImplIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/storage/log/SnapshotStoreImplIT.java
@@ -174,6 +174,15 @@ public class SnapshotStoreImplIT {
assertEquals(makeComparable(snapshot1), makeComparable(snapshot2));
}
+ @Test
+ public void testBackfill() {
+ setUpStore(false);
+ snapshotStore.applySnapshot(makeNonBackfilled());
+
+ Snapshot backfilled = snapshotStore.createSnapshot();
+ assertEquals(expected(), makeComparable(backfilled));
+ }
+
private static final IScheduledTask TASK = TaskTestUtil.makeTask("id", JOB_KEY);
private static final ITaskConfig TASK_CONFIG = TaskTestUtil.makeConfig(JOB_KEY);
private static final IJobConfiguration CRON_JOB = IJobConfiguration.build(new JobConfiguration()
@@ -244,6 +253,21 @@ public class SnapshotStoreImplIT {
new StoredJobUpdateDetails(UPDATE.newBuilder(), LOCK.getToken())));
}
+ private Snapshot makeNonBackfilled() {
+ Snapshot snapshot = expected();
+ snapshot.getTasks().forEach(e -> e.getAssignedTask().getTask().unsetResources());
+ snapshot.getCronJobs()
+ .forEach(e -> e.getJobConfiguration().getTaskConfig().unsetResources());
+ snapshot.getJobUpdateDetails()
+ .forEach(e -> e.getDetails().getUpdate().getInstructions()
+ .getDesiredState().getTask().unsetResources());
+ snapshot.getJobUpdateDetails()
+ .forEach(e -> e.getDetails().getUpdate().getInstructions()
+ .getInitialState().forEach(i -> i.getTask().unsetResources()));
+
+ return snapshot;
+ }
+
private void populateStore() {
storage.write((NoResult.Quiet) store -> {
store.getUnsafeTaskStore().saveTasks(ImmutableSet.of(TASK));
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/storage/log/ThriftBackfillTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/storage/log/ThriftBackfillTest.java b/src/test/java/org/apache/aurora/scheduler/storage/log/ThriftBackfillTest.java
new file mode 100644
index 0000000..cc039fb
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/storage/log/ThriftBackfillTest.java
@@ -0,0 +1,96 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.storage.log;
+
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.aurora.gen.TaskConfig;
+import org.junit.Test;
+
+import static org.apache.aurora.gen.Resource.diskMb;
+import static org.apache.aurora.gen.Resource.namedPort;
+import static org.apache.aurora.gen.Resource.numCpus;
+import static org.apache.aurora.gen.Resource.ramMb;
+import static org.junit.Assert.assertEquals;
+
+public class ThriftBackfillTest {
+
+ @Test
+ public void testFieldsToSetNoPorts() {
+ TaskConfig config = new TaskConfig()
+ .setNumCpus(1.0)
+ .setRamMb(32)
+ .setDiskMb(64);
+
+ TaskConfig expected = config.deepCopy();
+ expected.setResources(ImmutableSet.of(numCpus(1.0), ramMb(32), diskMb(64)));
+
+ assertEquals(
+ expected,
+ ThriftBackfill.backfillTask(config));
+ }
+
+ @Test
+ public void testFieldsToSetWithPorts() {
+ TaskConfig config = new TaskConfig()
+ .setNumCpus(1.0)
+ .setRamMb(32)
+ .setDiskMb(64)
+ .setRequestedPorts(ImmutableSet.of("http"));
+
+ TaskConfig expected = config.deepCopy();
+ expected.setResources(ImmutableSet.of(numCpus(1.0), ramMb(32), diskMb(64), namedPort("http")));
+
+ assertEquals(
+ expected,
+ ThriftBackfill.backfillTask(config));
+ }
+
+ @Test
+ public void testSetToFieldsNoPorts() {
+ TaskConfig config = new TaskConfig()
+ .setResources(ImmutableSet.of(numCpus(1.0), ramMb(32), diskMb(64)));
+
+ TaskConfig expected = config.deepCopy()
+ .setNumCpus(1.0)
+ .setRamMb(32)
+ .setDiskMb(64);
+
+ assertEquals(
+ expected,
+ ThriftBackfill.backfillTask(config));
+ }
+
+ @Test
+ public void testSetToFieldsWithPorts() {
+ TaskConfig config = new TaskConfig()
+ .setResources(ImmutableSet.of(numCpus(1.0), ramMb(32), diskMb(64), namedPort("http")));
+
+ TaskConfig expected = config.deepCopy()
+ .setNumCpus(1.0)
+ .setRamMb(32)
+ .setDiskMb(64)
+ .setRequestedPorts(ImmutableSet.of("http"));
+
+ assertEquals(
+ expected,
+ ThriftBackfill.backfillTask(config));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testMissingResourceThrows() {
+ TaskConfig config = new TaskConfig().setResources(ImmutableSet.of(numCpus(1.0), ramMb(32)));
+ ThriftBackfill.backfillTask(config);
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/thrift/Fixtures.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/Fixtures.java b/src/test/java/org/apache/aurora/scheduler/thrift/Fixtures.java
index fd6a40c..54585a9 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/Fixtures.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/Fixtures.java
@@ -36,6 +36,7 @@ import org.apache.aurora.gen.JobSummaryResult;
import org.apache.aurora.gen.JobUpdateKey;
import org.apache.aurora.gen.LockKey;
import org.apache.aurora.gen.MesosContainer;
+import org.apache.aurora.gen.Resource;
import org.apache.aurora.gen.ResourceAggregate;
import org.apache.aurora.gen.Response;
import org.apache.aurora.gen.ResponseCode;
@@ -111,7 +112,11 @@ final class Fixtures {
.setMaxTaskFailures(1)
.setConstraints(ImmutableSet.of())
.setMetadata(ImmutableSet.of())
- .setContainer(Container.mesos(new MesosContainer()));
+ .setContainer(Container.mesos(new MesosContainer()))
+ .setResources(ImmutableSet.of(
+ Resource.numCpus(1),
+ Resource.ramMb(1024),
+ Resource.diskMb(1024)));
}
static TaskConfig nonProductionTask() {
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImplTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImplTest.java b/src/test/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImplTest.java
index 50a1fdb..3a2b3f3 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImplTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImplTest.java
@@ -705,11 +705,11 @@ public class ReadOnlySchedulerImplTest extends EasyMockTest {
@Test
public void testGetJobUpdateDiffWithUpdateAdd() throws Exception {
- TaskConfig task1 = defaultTask(false).setNumCpus(1.0);
- TaskConfig task2 = defaultTask(false).setNumCpus(2.0);
- TaskConfig task3 = defaultTask(false).setNumCpus(3.0);
- TaskConfig task4 = defaultTask(false).setNumCpus(4.0);
- TaskConfig task5 = defaultTask(false).setNumCpus(5.0);
+ TaskConfig task1 = defaultTask(false).setMaxTaskFailures(1);
+ TaskConfig task2 = defaultTask(false).setMaxTaskFailures(2);
+ TaskConfig task3 = defaultTask(false).setMaxTaskFailures(3);
+ TaskConfig task4 = defaultTask(false).setMaxTaskFailures(4);
+ TaskConfig task5 = defaultTask(false).setMaxTaskFailures(5);
ImmutableSet.Builder<IScheduledTask> tasks = ImmutableSet.builder();
makeTasks(0, 10, task1, tasks);
@@ -723,7 +723,7 @@ public class ReadOnlySchedulerImplTest extends EasyMockTest {
control.replay();
- TaskConfig newTask = defaultTask(false).setNumCpus(6.0);
+ TaskConfig newTask = defaultTask(false).setMaxTaskFailures(6);
JobUpdateRequest request = new JobUpdateRequest()
.setTaskConfig(newTask)
.setInstanceCount(60)
@@ -746,9 +746,9 @@ public class ReadOnlySchedulerImplTest extends EasyMockTest {
@Test
public void testGetJobUpdateDiffWithUpdateRemove() throws Exception {
- TaskConfig task1 = defaultTask(false).setNumCpus(1.0);
- TaskConfig task2 = defaultTask(false).setNumCpus(2.0);
- TaskConfig task3 = defaultTask(false).setNumCpus(3.0);
+ TaskConfig task1 = defaultTask(false).setMaxTaskFailures(1);
+ TaskConfig task2 = defaultTask(false).setMaxTaskFailures(2);
+ TaskConfig task3 = defaultTask(false).setMaxTaskFailures(3);
ImmutableSet.Builder<IScheduledTask> tasks = ImmutableSet.builder();
makeTasks(0, 10, task1, tasks);
@@ -761,7 +761,7 @@ public class ReadOnlySchedulerImplTest extends EasyMockTest {
control.replay();
JobUpdateRequest request = new JobUpdateRequest()
- .setTaskConfig(defaultTask(false).setNumCpus(6.0))
+ .setTaskConfig(defaultTask(false).setMaxTaskFailures(6))
.setInstanceCount(20)
.setSettings(new JobUpdateSettings());
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java b/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
index e6e79a0..acc68ae 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
@@ -59,6 +59,7 @@ import org.apache.aurora.gen.PulseJobUpdateResult;
import org.apache.aurora.gen.QueryRecoveryResult;
import org.apache.aurora.gen.Range;
import org.apache.aurora.gen.ReadOnlyScheduler;
+import org.apache.aurora.gen.Resource;
import org.apache.aurora.gen.ResourceAggregate;
import org.apache.aurora.gen.Response;
import org.apache.aurora.gen.ResponseDetail;
@@ -504,7 +505,11 @@ public class SchedulerThriftInterfaceTest extends EasyMockTest {
.setRequestedPorts(ImmutableSet.of())
.setTaskLinks(ImmutableMap.of())
.setConstraints(ImmutableSet.of())
- .setMaxTaskFailures(0);
+ .setMaxTaskFailures(0)
+ .setResources(ImmutableSet.of(
+ Resource.numCpus(1.0),
+ Resource.ramMb(1024),
+ Resource.diskMb(1024)));
lockManager.assertNotLocked(LOCK_KEY);
storageUtil.expectTaskFetch(Query.jobScoped(JOB_KEY).active());
@@ -558,7 +563,6 @@ public class SchedulerThriftInterfaceTest extends EasyMockTest {
.setAssignedTask(new AssignedTask()
.setInstanceId(instanceId)
.setTask(populatedTask()
- .setRamMb(5)
.setIsService(true)
.setExecutorConfig(new ExecutorConfig().setData(executorData)))));
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/ff6e05f0/src/test/python/apache/aurora/client/cli/test_inspect.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/client/cli/test_inspect.py b/src/test/python/apache/aurora/client/cli/test_inspect.py
index 452699c..fedc16b 100644
--- a/src/test/python/apache/aurora/client/cli/test_inspect.py
+++ b/src/test/python/apache/aurora/client/cli/test_inspect.py
@@ -89,7 +89,10 @@ Process 'process':
cmd = AuroraCommandLine()
assert cmd.execute(['job', 'inspect', '--raw', 'west/bozo/test/hello', 'config.aurora']) == 0
output = '\n'.join(mock_stdout)
- assert output == str(job_config.job())
+ # It's impossible to assert string equivalence of two objects with nested un-hashable types.
+ # Given that the only product of --raw flag is the thrift representation of AuroraConfig
+ # it's enough to do a spot check here and let thrift.py tests validate the structure.
+ assert 'TaskConfig' in output
# AURORA-990: Prevent regression of client passing invalid arguments to print_out.
# Since print_out is the final layer before print(), there's not much else we can do than