Skip to content

KAFKA-1683: persisting session information in Requests #155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gwenshap wants to merge 6 commits into from
Closed

KAFKA-1683: persisting session information in Requests #155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d97449d
persisting session information in Requests
gwenshap Aug 21, 2015
e55c50b
adding logging per Harsha's advice
gwenshap Aug 21, 2015
d4d4ed7
avoiding hardcoded ANONYMOUS string everywhere, per Parth's comment
gwenshap Aug 21, 2015
ba8a334
addressing suggestions by Ismael and Edward
gwenshap Aug 21, 2015
8d40258
added security protocol and principal to request logs
gwenshap Aug 24, 2015
48821bf
minor fix for logs following Jun's comments
gwenshap Aug 26, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/network/PlaintextTransportLayer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public class PlaintextTransportLayer implements TransportLayer {
private static final Logger log = LoggerFactory.getLogger(PlaintextTransportLayer.class);
private final SelectionKey key;
private final SocketChannel socketChannel;
private final Principal principal = new KafkaPrincipal("ANONYMOUS");
private final Principal principal = KafkaPrincipal.ANONYMOUS;

public PlaintextTransportLayer(SelectionKey key) throws IOException {
this.key = key;
Expand Down
4 changes: 3 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLPeerUnverifiedException;

import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -600,7 +601,8 @@ public Principal peerPrincipal() throws IOException {
try {
return sslEngine.getSession().getPeerPrincipal();
} catch (SSLPeerUnverifiedException se) {

@harshach harshach Aug 21, 2015

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we log a warning here

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah, since we are caching the Principal in the authenticator, warning here makes sense.

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

btw. this is pretty weird - the session and the tests are on the server side, but since we are using the client network layer, you need to have log4j.logger.org.apache.kafka=WARN to see this warning.

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just for my benefit, what would this warning indicate to a Kafka user? Can they do something about it?

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will indicate that even though they are using SSL, it is SSL without authentication (just encryption). It could be intentional, but could also be a misconfiguration (such as misplaced key files), hence warning.

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Understood, thanks.

throw new IOException(String.format("Unable to retrieve getPeerPrincipal due to %s", se));
log.warn("SSL peer is not authenticated, returning ANONYMOUS instead");
return KafkaPrincipal.ANONYMOUS;
}
}

Expand Down
7 changes: 5 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/network/Selector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,6 +465,7 @@ private void close(KafkaChannel channel) {
this.sensors.connectionClosed.record();
}


/**
* check if channel is ready
*/
Expand All @@ -475,9 +476,11 @@ public boolean isChannelReady(String id) {
}

/**
* Get the channel associated with this numeric id
* Get the channel associated with this connection
* Exposing this to allow SocketServer get the Principal from the channel when creating a request
* without making Selector know about Principals
*/
private KafkaChannel channelForId(String id) {
public KafkaChannel channelForId(String id) {
KafkaChannel channel = this.channels.get(id);
if (channel == null)
throw new IllegalStateException("Attempt to write to socket for which there is no open connection. Connection id " + id + " existing connections " + channels.keySet().toString());
Expand Down
1 change: 1 addition & 0 deletions clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.security.Principal;

public class KafkaPrincipal implements Principal {
public final static KafkaPrincipal ANONYMOUS = new KafkaPrincipal("ANONYMOUS");
private final String name;

public KafkaPrincipal(String name) {
Expand Down
16 changes: 10 additions & 6 deletions core/src/main/scala/kafka/network/RequestChannel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package kafka.network

import java.nio.ByteBuffer
import java.security.Principal
import java.util.concurrent._

import com.yammer.metrics.core.Gauge
Expand All @@ -29,11 +30,12 @@ import kafka.utils.{Logging, SystemTime}
import org.apache.kafka.common.network.Send
import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol}
import org.apache.kafka.common.requests.{AbstractRequest, RequestHeader}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.log4j.Logger


object RequestChannel extends Logging {
val AllDone = new Request(processor = 1, connectionId = "2", buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT)
val AllDone = new Request(processor = 1, connectionId = "2", new Session(KafkaPrincipal.ANONYMOUS, ""), buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT)

@junrao junrao Aug 23, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It will be useful to include the principal and the security type when logging the request in the requestLogger. Also, in the request log, instead of logging connectionId, we should log session.host instead.

@gwenshap gwenshap Aug 24, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added the principal and security protocol to the logging. I left connectionId in place - currently it contains the host:port pairs of both sides of the connection and is therefore very informative. I'd like to keep this level of detail.

def getShutdownReceive() = {
val emptyProducerRequest = new ProducerRequest(0, 0, "", 0, 0, collection.mutable.Map[TopicAndPartition, ByteBufferMessageSet]())
Expand All @@ -44,7 +46,9 @@ object RequestChannel extends Logging {
byteBuffer
}

case class Request(processor: Int, connectionId: String, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {
case class Session(principal: Principal, host: String)

@eribeiro eribeiro Aug 21, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Session would have an id: Int to identify it, right? I mean, in the context of various "anonymous" principal connecting to the same host we need a way of uniquely identifying each session. Or would this id be the same as connectionId (i.e., no multiplexing multi-sessions on a single connection)? I don't know. wdyt?

update: I can see some methods/fields related to valid session state (valid, isValid, invalidate(), etc), but those can come in time, I think. :)

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe the name is bad, but we didn't really plan on session IDs.
Sessions are always 1-1 with connections and those have the ID. The ID is srcHost:srcPort:destHost:destPort - which is guaranteed to be unique.

@eribeiro eribeiro Aug 21, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, nice. :) 👍 Yeah... maybe ConnectionState or ConnectionInfo could be a better name. Just a suggestion.


case class Request(processor: Int, connectionId: String, session: Session, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {
@volatile var requestDequeueTimeMs = -1L
@volatile var apiLocalCompleteTimeMs = -1L
@volatile var responseCompleteTimeMs = -1L
Expand Down Expand Up @@ -113,11 +117,11 @@ object RequestChannel extends Logging {
}

if(requestLogger.isTraceEnabled)
requestLogger.trace("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d"
.format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime))
requestLogger.trace("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d,securityProtocol:%s,principal:%s"

@junrao junrao Aug 26, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To be consistent with the logging in KafkaApis, securityProtocol should be "security protocol"? Ditto for the logging below.

@ijuma ijuma Aug 26, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even though the same logging line includes totalTime, requestQueueTime and so on?

@gwenshap gwenshap Aug 26, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I tried to match the existing style here. I'll modify the extra details in KafkaApis to match.

@gwenshap gwenshap Aug 26, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here's what we have after my change, nice and consistent :)

[2015-08-26 11:33:06,280] TRACE [KafkaApi-0] Handling request:Name: TopicMetadataRequest; Version: 0; CorrelationId: 0; ClientId: console-producer; Topics: t1 from connection 127.0.0.1:9092-127.0.0.1:64559;securityProtocol:PLAINTEXT,principal:ANONYMOUS (kafka.server.KafkaApis)

[2015-08-26 11:33:06,280] TRACE Completed request:Name: TopicMetadataRequest; Version: 0; CorrelationId: 0; ClientId: console-producer from connection 127.0.0.1:9092-127.0.0.1:64559;totalTime:1,requestQueueTime:1,localTime:0,remoteTime:0,responseQueueTime:0,sendTime:0,securityProtocol:PLAINTEXT,principal:ANONYMOUS (kafka.request.logger)

.format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime, securityProtocol, session.principal))
else if(requestLogger.isDebugEnabled)
requestLogger.debug("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d"
.format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime))
requestLogger.debug("Completed request:%s from connection %s;totalTime:%d,requestQueueTime:%d,localTime:%d,remoteTime:%d,responseQueueTime:%d,sendTime:%d,securityProtocol:%s,principal:%s"
.format(requestDesc, connectionId, totalTime, requestQueueTime, apiLocalTime, apiRemoteTime, responseQueueTime, responseSendTime, securityProtocol, session.principal))
}
}

Expand Down
5 changes: 4 additions & 1 deletion core/src/main/scala/kafka/network/SocketServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,10 @@ private[kafka] class Processor(val id: Int,
}
collection.JavaConversions.collectionAsScalaIterable(selector.completedReceives).foreach(receive => {
try {
val req = RequestChannel.Request(processor = id, connectionId = receive.source, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)

val channel = selector.channelForId(receive.source);
val session = RequestChannel.Session(channel.principal, channel.socketDescription)
val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)
requestChannel.sendRequest(req)
} catch {
case e @ (_: InvalidRequestException | _: SchemaException) => {
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@ class KafkaApis(val requestChannel: RequestChannel,
*/
def handle(request: RequestChannel.Request) {
try{
trace("Handling request: " + request.requestObj + " from connection: " + request.connectionId)
trace("Handling request:%s from connection %s;securityProtocol:%s,principal:%s".
format(request.requestObj, request.connectionId, request.securityProtocol, request.session.principal))
request.requestId match {
case RequestKeys.ProduceKey => handleProducerRequest(request)
case RequestKeys.FetchKey => handleFetchRequest(request)
Expand Down
19 changes: 10 additions & 9 deletions core/src/test/scala/unit/kafka/network/SocketServerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,21 +17,15 @@

package kafka.network;

import java.io._

import java.net._
import javax.net.ssl._
import java.io._
import java.nio.ByteBuffer
import java.util.Random

import kafka.api.ProducerRequest
import kafka.cluster.EndPoint
import kafka.common.TopicAndPartition
import kafka.message.ByteBufferMessageSet
import kafka.producer.SyncProducerConfig
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.NetworkSend
import org.apache.kafka.common.protocol.SecurityProtocol
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.SystemTime
import org.junit.Assert._
import org.junit._
Expand All @@ -43,7 +37,6 @@ import java.nio.ByteBuffer
import kafka.common.TopicAndPartition
import kafka.message.ByteBufferMessageSet
import kafka.server.KafkaConfig
import java.nio.channels.SelectionKey
import kafka.utils.TestUtils

import scala.collection.Map
Expand Down Expand Up @@ -230,4 +223,12 @@ class SocketServerTest extends JUnitSuite {
assertEquals(serializedBytes.toSeq, receiveResponse(sslSocket).toSeq)
overrideServer.shutdown()
}

@Test
def testSessionPrincipal(): Unit = {
val socket = connect()
val bytes = new Array[Byte](40)
sendRequest(socket, 0, bytes)
assertEquals(KafkaPrincipal.ANONYMOUS, server.requestChannel.receiveRequest().session.principal)
}
}