Stop making sensitive kafka information available for logging (knativ…

…e#329)

* Stop making sensitive kafka information available for logging

The adapter currently logs the kafka message directly to the console
instead; log metadata such as partition, topic, offset instead
The receive adapter is currently even worse due to the zap logging
done by reflection.  This prints out every struct parameter including
user/password combo.  Drop that and log less sensitive metadata.

* Fix Apache Kafka logged adapter metadata to be consistent

* Fix formatting, casting of logging messages

- receive_adapter/main.go - remove string casts
                          - change 'Bootstap Server' to 'bootstrapServer'
- adapter.go - remove casts and change to proper zap.Int{32,64} functions
             - change logging level to Debug (from Info)

* Logging format change: bootstrapServer to bootstrap_server

contrib/kafka/cmd/receive_adapter/main.go

@@ -94,7 +94,12 @@ func main() {
 
 	stopCh := signals.SetupSignalHandler()
 
-	logger.Info("Starting Apache Kafka Receive Adapter...", zap.Reflect("adapter", adapter))
+	logger.Info("Starting Apache Kafka Receive Adapter...",
+		zap.String("bootstrap_server", adapter.BootstrapServers),
+		zap.String("Topics", adapter.Topics),
+		zap.String("ConsumerGroup", adapter.ConsumerGroup),
+		zap.String("SinkURI", adapter.SinkURI),
+		zap.Bool("TLS", adapter.Net.SASL.Enable))
 	if err := adapter.Start(ctx, stopCh); err != nil {
 		logger.Fatal("failed to start adapter: ", zap.Error(err))
 	}

contrib/kafka/pkg/adapter/adapter.go

@@ -77,7 +77,9 @@ func (a *Adapter) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.Co
 	logger := logging.FromContext(context.TODO())
 
 	for msg := range claim.Messages() {
-		logger.Info("Received: ", zap.Any("value", string(msg.Value)))
+		logger.Debug("Received: ", zap.String("topic:", msg.Topic),
+			zap.Int32("partition:", msg.Partition),
+			zap.Int64("offset:", msg.Offset))
 
 		// send and mark message if post was successful
 		if err := a.postMessage(context.TODO(), msg); err == nil {

contrib/kafka/samples/README.md

@@ -103,7 +103,7 @@ and an Event Display Service.
    configuration.
    ```
    $ kubectl logs kafka-source-xlnhq-5544766765-dnl5s
-   {"level":"info","ts":"2019-03-19T22:31:52.689Z","caller":"receive_adapter/main.go:97","msg":"Starting Apache Kafka Receive Adapter...","adapter":{"BootstrapServers":"...","Topic":"...","ConsumerGroup":"...","Net":{"SASL":{"Enable":true,"User":"...","Password":"..."},"TLS":{"Enable":true}},"SinkURI":"http://event-display.default.svc.cluster.local/"}}
+   {"level":"info","ts":"2019-04-01T19:09:32.164Z","caller":"receive_adapter/main.go:97","msg":"Starting Apache Kafka Receive Adapter...","Bootstrap Server":"...","Topics":".","ConsumerGroup":"...","SinkURI":"http://event-display.default.svc.cluster.local/","TLS":false}
    ```
 
 ### Verify
@@ -117,7 +117,7 @@ and an Event Display Service.
    ```
    $ kubectl logs kafka-source-xlnhq-5544766765-dnl5s
    ...
-   {"level":"info","ts":1553034726.5351956,"logger":"fallback","caller":"adapter/adapter.go:121","msg":"Received: {value 15 0 {\"msg\": \"This is a test!\"} <nil>}"}
+   {"level":"info","ts":1554145778.9344022,"logger":"fallback","caller":"adapter/adapter.go:80","msg":"Received: {topic: 15 0 ... <nil>} {partition: 11 2  <nil>} {offset: 11 0  <nil>}"}
    {"level":"info","ts":1553034726.546107,"logger":"fallback","caller":"adapter/adapter.go:154","msg":"Successfully sent event to sink"}
    ```
 3. Ensure the Event Display received the message sent to it by the Event Source.