Skip to content

Commit

Permalink
Expose authorization state in transform stats
Browse files Browse the repository at this point in the history
  • Loading branch information
przemekwitek committed Mar 29, 2023
1 parent 56d53da commit f94580f
Show file tree
Hide file tree
Showing 40 changed files with 1,414 additions and 222 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

package org.elasticsearch.xpack.core.transform.action;

import org.elasticsearch.TransportVersion;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.support.tasks.BaseTasksRequest;
Expand All @@ -19,6 +20,7 @@
import org.elasticsearch.xcontent.XContentParser;
import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
import org.elasticsearch.xpack.core.transform.TransformField;
import org.elasticsearch.xpack.core.transform.transforms.AuthorizationState;
import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
import org.elasticsearch.xpack.core.transform.transforms.TransformConfigUpdate;

Expand Down Expand Up @@ -46,6 +48,7 @@ public static class Request extends BaseTasksRequest<Request> {
private final String id;
private final boolean deferValidation;
private TransformConfig config;
private AuthorizationState authState;

public Request(TransformConfigUpdate update, String id, boolean deferValidation, TimeValue timeout) {
this.update = update;
Expand All @@ -62,6 +65,11 @@ public Request(StreamInput in) throws IOException {
if (in.readBoolean()) {
this.config = new TransformConfig(in);
}
if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0)) {
if (in.readBoolean()) {
this.authState = new AuthorizationState(in);
}
}
}

public static Request fromXContent(
Expand Down Expand Up @@ -124,6 +132,14 @@ public void setConfig(TransformConfig config) {
this.config = config;
}

public AuthorizationState getAuthState() {
return authState;
}

public void setAuthState(AuthorizationState authState) {
this.authState = authState;
}

@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
Expand All @@ -136,12 +152,20 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeBoolean(true);
config.writeTo(out);
}
if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0)) {
if (authState == null) {
out.writeBoolean(false);
} else {
out.writeBoolean(true);
authState.writeTo(out);
}
}
}

@Override
public int hashCode() {
// the base class does not implement hashCode, therefore we need to hash timeout ourselves
return Objects.hash(getTimeout(), update, id, deferValidation, config);
return Objects.hash(getTimeout(), update, id, deferValidation, config, authState);
}

@Override
Expand All @@ -159,6 +183,7 @@ public boolean equals(Object obj) {
&& this.deferValidation == other.deferValidation
&& this.id.equals(other.id)
&& Objects.equals(config, other.config)
&& Objects.equals(authState, other.authState)
&& getTimeout().equals(other.getTimeout());
}
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License
* 2.0; you may not use this file except in compliance with the Elastic License
* 2.0.
*/

package org.elasticsearch.xpack.core.transform.transforms;

import org.elasticsearch.common.Strings;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.health.HealthStatus;
import org.elasticsearch.xcontent.ConstructingObjectParser;
import org.elasticsearch.xcontent.ObjectParser;
import org.elasticsearch.xcontent.ParseField;
import org.elasticsearch.xcontent.ToXContent;
import org.elasticsearch.xcontent.ToXContentObject;
import org.elasticsearch.xcontent.XContentBuilder;
import org.elasticsearch.xcontent.XContentParser;

import java.io.IOException;
import java.time.Instant;
import java.util.Locale;
import java.util.Objects;

/**
* {@link AuthorizationState} holds the state of the authorization performed in the past.
* By examining the instance of this class the caller can learn whether or not the user was authorized to access the source/dest indices
* present in the {@link TransformConfig}.
*
* This class is immutable.
*/
public class AuthorizationState implements Writeable, ToXContentObject {

public static AuthorizationState green() {
return new AuthorizationState(System.currentTimeMillis(), HealthStatus.GREEN, null);
}

public static AuthorizationState red(Exception e) {
return new AuthorizationState(System.currentTimeMillis(), HealthStatus.RED, e != null ? e.getMessage() : "unknown exception");
}

public static final ParseField TIMESTAMP = new ParseField("timestamp");
public static final ParseField STATUS = new ParseField("status");
public static final ParseField LAST_AUTH_ERROR = new ParseField("last_auth_error");

public static final ConstructingObjectParser<AuthorizationState, Void> PARSER = new ConstructingObjectParser<>(
"transform_authorization_state",
true,
a -> new AuthorizationState((Long) a[0], (HealthStatus) a[1], (String) a[2])
);

static {
PARSER.declareLong(ConstructingObjectParser.constructorArg(), TIMESTAMP);
PARSER.declareField(ConstructingObjectParser.constructorArg(), p -> {
if (p.currentToken() == XContentParser.Token.VALUE_STRING) {
return HealthStatus.valueOf(p.text().toUpperCase(Locale.ROOT));
}
throw new IllegalArgumentException("Unsupported token [" + p.currentToken() + "]");
}, STATUS, ObjectParser.ValueType.STRING);
PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), LAST_AUTH_ERROR);
}

private final long timestampMillis;
private final HealthStatus status;
@Nullable
private final String lastAuthError;

public AuthorizationState(Long timestamp, HealthStatus status, @Nullable String lastAuthError) {
this.timestampMillis = timestamp;
this.status = status;
this.lastAuthError = lastAuthError;
}

public AuthorizationState(StreamInput in) throws IOException {
this.timestampMillis = in.readLong();
this.status = in.readEnum(HealthStatus.class);
this.lastAuthError = in.readOptionalString();
}

public Instant getTimestamp() {
return Instant.ofEpochMilli(timestampMillis);
}

public HealthStatus getStatus() {
return status;
}

public String getLastAuthError() {
return lastAuthError;
}

@Override
public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
builder.startObject();
builder.field(TIMESTAMP.getPreferredName(), timestampMillis);
builder.field(STATUS.getPreferredName(), status.xContentValue());
if (lastAuthError != null) {
builder.field(LAST_AUTH_ERROR.getPreferredName(), lastAuthError);
}
builder.endObject();
return builder;
}

@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeLong(timestampMillis);
status.writeTo(out);
out.writeOptionalString(lastAuthError);
}

@Override
public boolean equals(Object other) {
if (this == other) {
return true;
}

if (other == null || getClass() != other.getClass()) {
return false;
}

AuthorizationState that = (AuthorizationState) other;

return this.timestampMillis == that.timestampMillis
&& this.status.value() == that.status.value()
&& Objects.equals(this.lastAuthError, that.lastAuthError);
}

@Override
public int hashCode() {
return Objects.hash(timestampMillis, status, lastAuthError);
}

@Override
public String toString() {
return Strings.toString(this);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,7 @@ public static TransformCheckpoint fromXContent(final XContentParser parser, bool

public static String documentId(String transformId, long checkpoint) {
if (checkpoint < 0) {
throw new IllegalArgumentException("checkpoint must be a positive number");
throw new IllegalArgumentException("checkpoint must be a non-negative number");
}

return NAME + "-" + transformId + "-" + checkpoint;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ public class TransformConfigUpdate implements Writeable {

public static final String NAME = "data_frame_transform_config_update";

public static TransformConfigUpdate EMPTY = new TransformConfigUpdate(null, null, null, null, null, null, null, null);
public static final TransformConfigUpdate EMPTY = new TransformConfigUpdate(null, null, null, null, null, null, null, null);

private static final ConstructingObjectParser<TransformConfigUpdate, String> PARSER = new ConstructingObjectParser<>(
NAME,
Expand Down Expand Up @@ -235,6 +235,10 @@ public boolean changesSettings(TransformConfig config) {
return isNullOrEqual(settings, config.getSettings()) == false;
}

public boolean changesHeaders(TransformConfig config) {
return isNullOrEqual(headers, config.getHeaders()) == false;
}

private boolean isNullOrEqual(Object lft, Object rgt) {
return lft == null || lft.equals(rgt);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ public class TransformState implements Task.Status, PersistentTaskState {
private NodeAttributes node;

private final boolean shouldStopAtNextCheckpoint;
@Nullable
private final AuthorizationState authState;

public static final ParseField TASK_STATE = new ParseField("task_state");
public static final ParseField INDEXER_STATE = new ParseField("indexer_state");
Expand All @@ -57,6 +59,7 @@ public class TransformState implements Task.Status, PersistentTaskState {
public static final ParseField PROGRESS = new ParseField("progress");
public static final ParseField NODE = new ParseField("node");
public static final ParseField SHOULD_STOP_AT_NEXT_CHECKPOINT = new ParseField("should_stop_at_checkpoint");
public static final ParseField AUTH_STATE = new ParseField("auth_state");

@SuppressWarnings("unchecked")
public static final ConstructingObjectParser<TransformState, Void> PARSER = new ConstructingObjectParser<>(NAME, true, args -> {
Expand All @@ -75,6 +78,7 @@ public class TransformState implements Task.Status, PersistentTaskState {
TransformProgress progress = (TransformProgress) args[6];
NodeAttributes node = (NodeAttributes) args[7];
boolean shouldStopAtNextCheckpoint = args[8] == null ? false : (boolean) args[8];
AuthorizationState authState = (AuthorizationState) args[9];

return new TransformState(
taskState,
Expand All @@ -84,7 +88,8 @@ public class TransformState implements Task.Status, PersistentTaskState {
reason,
progress,
node,
shouldStopAtNextCheckpoint
shouldStopAtNextCheckpoint,
authState
);
});

Expand All @@ -98,6 +103,7 @@ public class TransformState implements Task.Status, PersistentTaskState {
PARSER.declareField(optionalConstructorArg(), TransformProgress.PARSER::apply, PROGRESS, ValueType.OBJECT);
PARSER.declareField(optionalConstructorArg(), NodeAttributes.PARSER::apply, NODE, ValueType.OBJECT);
PARSER.declareBoolean(optionalConstructorArg(), SHOULD_STOP_AT_NEXT_CHECKPOINT);
PARSER.declareField(optionalConstructorArg(), AuthorizationState.PARSER::apply, AUTH_STATE, ValueType.OBJECT);
}

public TransformState(
Expand All @@ -108,7 +114,8 @@ public TransformState(
@Nullable String reason,
@Nullable TransformProgress progress,
@Nullable NodeAttributes node,
boolean shouldStopAtNextCheckpoint
boolean shouldStopAtNextCheckpoint,
@Nullable AuthorizationState authState
) {
this.taskState = taskState;
this.indexerState = indexerState;
Expand All @@ -118,29 +125,7 @@ public TransformState(
this.progress = progress;
this.node = node;
this.shouldStopAtNextCheckpoint = shouldStopAtNextCheckpoint;
}

public TransformState(
TransformTaskState taskState,
IndexerState indexerState,
@Nullable TransformIndexerPosition position,
long checkpoint,
@Nullable String reason,
@Nullable TransformProgress progress,
@Nullable NodeAttributes node
) {
this(taskState, indexerState, position, checkpoint, reason, progress, node, false);
}

public TransformState(
TransformTaskState taskState,
IndexerState indexerState,
@Nullable TransformIndexerPosition position,
long checkpoint,
@Nullable String reason,
@Nullable TransformProgress progress
) {
this(taskState, indexerState, position, checkpoint, reason, progress, null);
this.authState = authState;
}

public TransformState(StreamInput in) throws IOException {
Expand All @@ -165,6 +150,11 @@ public TransformState(StreamInput in) throws IOException {
} else {
shouldStopAtNextCheckpoint = false;
}
if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0)) {
authState = in.readOptionalWriteable(AuthorizationState::new);
} else {
authState = null;
}
}

public TransformTaskState getTaskState() {
Expand Down Expand Up @@ -204,6 +194,10 @@ public boolean shouldStopAtNextCheckpoint() {
return shouldStopAtNextCheckpoint;
}

public AuthorizationState getAuthState() {
return authState;
}

public static TransformState fromXContent(XContentParser parser) {
try {
return PARSER.parse(parser, null);
Expand Down Expand Up @@ -231,6 +225,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.field(NODE.getPreferredName(), node);
}
builder.field(SHOULD_STOP_AT_NEXT_CHECKPOINT.getPreferredName(), shouldStopAtNextCheckpoint);
if (authState != null) {
builder.field(AUTH_STATE.getPreferredName(), authState);
}
builder.endObject();
return builder;
}
Expand Down Expand Up @@ -258,6 +255,9 @@ public void writeTo(StreamOutput out) throws IOException {
if (out.getTransportVersion().onOrAfter(TransportVersion.V_7_6_0)) {
out.writeBoolean(shouldStopAtNextCheckpoint);
}
if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0)) {
out.writeOptionalWriteable(authState);
}
}

@Override
Expand All @@ -279,12 +279,13 @@ public boolean equals(Object other) {
&& Objects.equals(this.reason, that.reason)
&& Objects.equals(this.progress, that.progress)
&& Objects.equals(this.shouldStopAtNextCheckpoint, that.shouldStopAtNextCheckpoint)
&& Objects.equals(this.node, that.node);
&& Objects.equals(this.node, that.node)
&& Objects.equals(this.authState, that.authState);
}

@Override
public int hashCode() {
return Objects.hash(taskState, indexerState, position, checkpoint, reason, progress, node, shouldStopAtNextCheckpoint);
return Objects.hash(taskState, indexerState, position, checkpoint, reason, progress, node, shouldStopAtNextCheckpoint, authState);
}

@Override
Expand Down
Loading

0 comments on commit f94580f

Please sign in to comment.