Skip to content

Commit

Permalink
fix #4249 #4726: toning down logging of errors
Browse files Browse the repository at this point in the history
  • Loading branch information
@shawkins
shawkins committed Jan 4, 2023
1 parent 0bce8c4 commit a492858
Show file tree
Hide file tree
Showing 5 changed files with 44 additions and 40 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
### 6.4-SNAPSHOT

#### Bugs
* Fix #4249 #4726: prevent the over-logging of errors after the websocket has been closed

#### Improvements
* Fix #4637: all pod operations that require a ready / succeeded pod may use withReadyWaitTimeout, which supersedes withLogWaitTimeout.
Expand Down
28 changes: 16 additions & 12 deletions ...client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/ExecWebSocketListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ private void asyncWrite(WritableByteChannel channel, ByteBuffer b) {
if (closed.get()) {
LOGGER.debug("Stream write failed after close", t);
} else {
// This could happen if the user simply closes their stream prior to completion
// This could happen if the user simply closes their stream prior to completion
LOGGER.warn("Stream write failed", t);
}
}
Expand All @@ -211,7 +211,7 @@ private void asyncWrite(WritableByteChannel channel, ByteBuffer b) {
@Override
public void close() {
// simply sends a close, which will shut down the output
// it's expected that the server will respond with a close, but if not the input will be shutdown implicitly
// it's expected that the server will respond with a close, but if not the input will be shutdown implicitly
closeWebSocketOnce(1000, "Closing...");
}

Expand Down Expand Up @@ -280,18 +280,22 @@ public void onError(WebSocket webSocket, Throwable t) {
status.setMessage(t.getMessage());
cleanUpOnce();
} finally {
try {
if (listener != null) {
ExecListener.Response execResponse = null;
if (response != null) {
execResponse = new SimpleResponse(response);
if (exitCode.isDone()) {
LOGGER.debug("Exec failure after done", t);
} else {
try {
if (listener != null) {
ExecListener.Response execResponse = null;
if (response != null) {
execResponse = new SimpleResponse(response);
}
listener.onFailure(t, execResponse);
} else {
LOGGER.error("Exec Failure", t);
}
listener.onFailure(t, execResponse);
} else {
LOGGER.error("Exec Failure", t);
} finally {
exitCode.completeExceptionally(t);
}
} finally {
exitCode.completeExceptionally(t);
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion ...lient/src/main/java/io/fabric8/kubernetes/client/dsl/internal/PortForwarderWebsocket.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,7 @@ public void close() {
socket.cancel(true);
socket.whenComplete((w, t) -> {
if (w != null) {
w.sendClose(1001, "User closing");
listener.closeBothWays(w, 1001, "User closing");
}
});
}
Expand Down
48 changes: 23 additions & 25 deletions ...c/main/java/io/fabric8/kubernetes/client/dsl/internal/PortForwarderWebsocketListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@
*/
package io.fabric8.kubernetes.client.dsl.internal;

import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.http.WebSocket;
import io.fabric8.kubernetes.client.utils.Utils;
import io.fabric8.kubernetes.client.utils.internal.SerialExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -46,8 +48,6 @@ public class PortForwarderWebsocketListener implements WebSocket.Listener {

private final AtomicBoolean alive = new AtomicBoolean(true);

private final AtomicBoolean errorOccurred = new AtomicBoolean(false);

final Collection<Throwable> clientThrowables = new CopyOnWriteArrayList<>();

final Collection<Throwable> serverThrowables = new CopyOnWriteArrayList<>();
Expand Down Expand Up @@ -75,9 +75,9 @@ public void onOpen(final WebSocket webSocket) {
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
logger.debug("Error while writing client data");
if (alive.get()) {
clientThrowables.add(e);
logger.error("Error while writing client data");
closeBothWays(webSocket, 1001, "Client error");
}
}
Expand All @@ -101,21 +101,27 @@ public void onMessage(WebSocket webSocket, ByteBuffer buffer) {
}

if (!buffer.hasRemaining()) {
errorOccurred.set(true);
logger.error("Received an empty message");
KubernetesClientException e = new KubernetesClientException("Received an empty message");
serverThrowables.add(e);
logger.debug("Protocol error", e);
closeBothWays(webSocket, 1002, PROTOCOL_ERROR);
return;
}

byte channel = buffer.get();
if (channel < 0 || channel > 1) {
errorOccurred.set(true);
logger.error("Received a wrong channel from the remote socket: {}", channel);
KubernetesClientException e = new KubernetesClientException(
String.format("Received a wrong channel from the remote socket: %s", channel));
serverThrowables.add(e);
logger.debug("Protocol error", e);
closeBothWays(webSocket, 1002, PROTOCOL_ERROR);
} else if (channel == 1) {
// Error channel
errorOccurred.set(true);
logger.error("Received an error from the remote socket");
// TODO: read the error
KubernetesClientException e = new KubernetesClientException(
String.format("Received an error from the remote socket"));
serverThrowables.add(e);
logger.debug("Server error", e);
closeForwarder();
} else {
// Data
Expand All @@ -136,7 +142,7 @@ public void onMessage(WebSocket webSocket, ByteBuffer buffer) {
}
if (alive.get()) {
clientThrowables.add(e);
logger.error("Error while forwarding data to the client", e);
logger.debug("Error while forwarding data to the client", e);
closeBothWays(webSocket, 1002, PROTOCOL_ERROR);
}
}
Expand All @@ -155,10 +161,9 @@ public void onClose(WebSocket webSocket, int code, String reason) {

@Override
public void onError(WebSocket webSocket, Throwable t) {
logger.debug("{}: onFailure", LOG_PREFIX);
logger.debug("{}: Throwable received from websocket", LOG_PREFIX, t);
if (alive.get()) {
serverThrowables.add(t);
logger.error("{}: Throwable received from websocket", LOG_PREFIX, t);
closeForwarder();
}
}
Expand All @@ -168,7 +173,7 @@ boolean isAlive() {
}

boolean errorOccurred() {
return errorOccurred.get() || !clientThrowables.isEmpty() || !serverThrowables.isEmpty();
return !clientThrowables.isEmpty() || !serverThrowables.isEmpty();
}

Collection<Throwable> getClientThrowables() {
Expand All @@ -179,33 +184,25 @@ Collection<Throwable> getServerThrowables() {
return serverThrowables;
}

private void closeBothWays(WebSocket webSocket, int code, String message) {
void closeBothWays(WebSocket webSocket, int code, String message) {
logger.debug("{}: Closing with code {} and reason: {}", LOG_PREFIX, code, message);
alive.set(false);
try {
webSocket.sendClose(code, message);
} catch (Exception e) {
serverThrowables.add(e);
logger.error("Error while closing the websocket", e);
logger.debug("Error while closing the websocket", e);
}
closeForwarder();
}

private void closeForwarder() {
alive.set(false);
if (in != null) {
try {
in.close();
} catch (IOException e) {
logger.error("{}: Error while closing the client input channel", LOG_PREFIX, e);
}
Utils.closeQuietly(in);
}
if (out != null && out != in) {
try {
out.close();
} catch (IOException e) {
logger.error("{}: Error while closing the client output channel", LOG_PREFIX, e);
}
Utils.closeQuietly(out);
}
pumperService.shutdownNow();
serialExecutor.shutdownNow();
Expand All @@ -228,4 +225,5 @@ private static void pipe(ReadableByteChannel in, WebSocket webSocket, BooleanSup
}
} while (isAlive.getAsBoolean() && read >= 0);
}

}
5 changes: 3 additions & 2 deletions ...st/java/io/fabric8/kubernetes/client/dsl/internal/PortForwarderWebsocketListenerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ void onMessage_withEmptyMessage_shouldEndWithError() {
verify(webSocket, timeout(10_000)).sendClose(1002, "Protocol error");
assertThat(outputContent.toString()).isEmpty();
assertThat(listener.errorOccurred()).isTrue();
assertThat(listener.getServerThrowables()).isEmpty();
assertThat(listener.getServerThrowables()).isNotEmpty();
assertThat(in.isOpen()).isFalse();
assertThat(out.isOpen()).isFalse();
}
Expand Down Expand Up @@ -210,7 +210,8 @@ void onMessage_withWrongChannel_shouldLogAndEndWithError() {
verify(webSocket, timeout(10_000)).sendClose(1002, "Protocol error");
assertThat(outputContent.toString()).isEmpty();
assertThat(listener.errorOccurred()).isTrue();
verify(logger).error("Received a wrong channel from the remote socket: {}", (byte) 5);
assertThat(listener.getServerThrowables().iterator().next().getMessage())
.isEqualTo("Received a wrong channel from the remote socket: 5");
}

}
Expand Down

0 comments on commit a492858

Please sign in to comment.