Hook configuration manager into solo broadcast

With the completion of the broadcast filtering framework, the
configuration manager, the policy manager, and the genesis block
creation, it is now possible for the orderer to filter and apply
configuration received over the wire as a transaction.  This changeset
adds that functionality for the solo ordering service.

https://jira.hyperledger.org/browse/FAB-593

Change-Id: I4e50c1f811d8cff02f67de7de278fbffe230f882
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

orderer/common/broadcastfilter/configfilter/configfilter.go

@@ -0,0 +1,52 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+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 configfilter
+
+import (
+	ab "github.com/hyperledger/fabric/orderer/atomicbroadcast"
+	"github.com/hyperledger/fabric/orderer/common/broadcastfilter"
+	"github.com/hyperledger/fabric/orderer/common/configtx"
+
+	"github.com/golang/protobuf/proto"
+)
+
+type configFilter struct {
+	configManager configtx.Manager
+}
+
+func New(manager configtx.Manager) broadcastfilter.Rule {
+	return &configFilter{
+		configManager: manager,
+	}
+}
+
+// Apply applies the rule to the given BroadcastMessage, replying with the Action to take for the message
+func (cf *configFilter) Apply(message *ab.BroadcastMessage) broadcastfilter.Action {
+	config := &ab.ConfigurationEnvelope{}
+
+	err := proto.Unmarshal(message.Data, config)
+	if err != nil {
+		return broadcastfilter.Forward
+	}
+
+	err = cf.configManager.Validate(config)
+	if err != nil {
+		return broadcastfilter.Reject
+	}
+
+	return broadcastfilter.Reconfigure
+}

orderer/common/broadcastfilter/configfilter/configfilter_test.go

@@ -0,0 +1,73 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+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 configfilter
+
+import (
+	"fmt"
+	"testing"
+
+	ab "github.com/hyperledger/fabric/orderer/atomicbroadcast"
+	"github.com/hyperledger/fabric/orderer/common/broadcastfilter"
+
+	"github.com/golang/protobuf/proto"
+)
+
+type mockConfigManager struct {
+	err error
+}
+
+func (mcm *mockConfigManager) Validate(configtx *ab.ConfigurationEnvelope) error {
+	return mcm.err
+}
+
+func (mcm *mockConfigManager) Apply(configtx *ab.ConfigurationEnvelope) error {
+	return mcm.err
+}
+
+func TestForwardNonConfig(t *testing.T) {
+	cf := New(&mockConfigManager{})
+	result := cf.Apply(&ab.BroadcastMessage{
+		Data: []byte("Opaque"),
+	})
+	if result != broadcastfilter.Forward {
+		t.Fatalf("Should have forwarded opaque message")
+	}
+}
+
+func TestAcceptGoodConfig(t *testing.T) {
+	cf := New(&mockConfigManager{})
+	config := &ab.ConfigurationEnvelope{}
+	configBytes, _ := proto.Marshal(config)
+	result := cf.Apply(&ab.BroadcastMessage{
+		Data: configBytes,
+	})
+	if result != broadcastfilter.Reconfigure {
+		t.Fatalf("Should have indiated a good config message causes a reconfiguration")
+	}
+}
+
+func TestRejectBadConfig(t *testing.T) {
+	cf := New(&mockConfigManager{err: fmt.Errorf("Error")})
+	config := &ab.ConfigurationEnvelope{}
+	configBytes, _ := proto.Marshal(config)
+	result := cf.Apply(&ab.BroadcastMessage{
+		Data: configBytes,
+	})
+	if result != broadcastfilter.Reject {
+		t.Fatalf("Should have rejected bad config message")
+	}
+}

orderer/main.go

@@ -28,6 +28,8 @@ import (
 	ab "github.com/hyperledger/fabric/orderer/atomicbroadcast"
 	"github.com/hyperledger/fabric/orderer/common/bootstrap"
 	"github.com/hyperledger/fabric/orderer/common/bootstrap/static"
+	"github.com/hyperledger/fabric/orderer/common/broadcastfilter"
+	"github.com/hyperledger/fabric/orderer/common/broadcastfilter/configfilter"
 	"github.com/hyperledger/fabric/orderer/common/configtx"
 	"github.com/hyperledger/fabric/orderer/common/policies"
 	"github.com/hyperledger/fabric/orderer/config"
@@ -119,6 +121,14 @@ func bootstrapConfigManager(lastConfigTx *ab.ConfigurationEnvelope) configtx.Man
 	return configManager
 }
 
+func createBroadcastRuleset(configManager configtx.Manager) *broadcastfilter.RuleSet {
+	return broadcastfilter.NewRuleSet([]broadcastfilter.Rule{
+		broadcastfilter.EmptyRejectRule,
+		configfilter.New(configManager),
+		broadcastfilter.AcceptRule,
+	})
+}
+
 func launchSolo(conf *config.TopLevel) {
 	grpcServer := grpc.NewServer()
 
@@ -175,7 +185,15 @@ func launchSolo(conf *config.TopLevel) {
 	// XXX actually use the config manager in the future
 	_ = configManager
 
-	solo.New(int(conf.General.QueueSize), int(conf.General.BatchSize), int(conf.General.MaxWindowSize), conf.General.BatchTimeout, rawledger, grpcServer)
+	solo.New(int(conf.General.QueueSize),
+		int(conf.General.BatchSize),
+		int(conf.General.MaxWindowSize),
+		conf.General.BatchTimeout,
+		rawledger,
+		grpcServer,
+		createBroadcastRuleset(configManager),
+		configManager,
+	)
 	grpcServer.Serve(lis)
 }
 

orderer/sample_clients/broadcast_timestamp/client.go

@@ -39,6 +39,17 @@ func (s *broadcastClient) broadcast(transaction []byte) error {
 	return s.client.Send(&ab.BroadcastMessage{Data: transaction})
 }
 
+func (s *broadcastClient) getAck() error {
+	msg, err := s.client.Recv()
+	if err != nil {
+		return err
+	}
+	if msg.Status != ab.Status_SUCCESS {
+		return fmt.Errorf("Got unexpected status: %v", msg.Status)
+	}
+	return nil
+}
+
 func main() {
 	config := config.Load()
 	serverAddr := fmt.Sprintf("%s:%d", config.General.ListenAddress, config.General.ListenPort)
@@ -56,4 +67,8 @@ func main() {
 
 	s := newBroadcastClient(client)
 	s.broadcast([]byte(fmt.Sprintf("Testing %v", time.Now())))
+	err = s.getAck()
+	if err != nil {
+		fmt.Printf("\n\n!!!!!!!!!!!!!!!!!\n%v\n!!!!!!!!!!!!!!!!!\n", err)
+	}
 }

orderer/sample_clients/deliver_stdout/client.go

@@ -73,6 +73,7 @@ func (r *deliverClient) readUntilClose() {
 	for {
 		msg, err := r.client.Recv()
 		if err != nil {
+			fmt.Println("Error receiving:", err)
 			return
 		}
 

orderer/solo/broadcast.go

@@ -21,34 +21,39 @@ import (
 
 	ab "github.com/hyperledger/fabric/orderer/atomicbroadcast"
 	"github.com/hyperledger/fabric/orderer/common/broadcastfilter"
+	"github.com/hyperledger/fabric/orderer/common/configtx"
 	"github.com/hyperledger/fabric/orderer/rawledger"
+
+	"github.com/golang/protobuf/proto"
 )
 
 type broadcastServer struct {
-	queueSize    int
-	batchSize    int
-	batchTimeout time.Duration
-	rl           rawledger.Writer
-	filter       *broadcastfilter.RuleSet
-	sendChan     chan *ab.BroadcastMessage
-	exitChan     chan struct{}
+	queueSize     int
+	batchSize     int
+	batchTimeout  time.Duration
+	rl            rawledger.Writer
+	filter        *broadcastfilter.RuleSet
+	configManager configtx.Manager
+	sendChan      chan *ab.BroadcastMessage
+	exitChan      chan struct{}
 }
 
-func newBroadcastServer(queueSize, batchSize int, batchTimeout time.Duration, rl rawledger.Writer) *broadcastServer {
-	bs := newPlainBroadcastServer(queueSize, batchSize, batchTimeout, rl)
+func newBroadcastServer(queueSize, batchSize int, batchTimeout time.Duration, rl rawledger.Writer, filters *broadcastfilter.RuleSet, configManager configtx.Manager) *broadcastServer {
+	bs := newPlainBroadcastServer(queueSize, batchSize, batchTimeout, rl, filters, configManager)
 	go bs.main()
 	return bs
 }
 
-func newPlainBroadcastServer(queueSize, batchSize int, batchTimeout time.Duration, rl rawledger.Writer) *broadcastServer {
+func newPlainBroadcastServer(queueSize, batchSize int, batchTimeout time.Duration, rl rawledger.Writer, filters *broadcastfilter.RuleSet, configManager configtx.Manager) *broadcastServer {
 	bs := &broadcastServer{
-		queueSize:    queueSize,
-		batchSize:    batchSize,
-		batchTimeout: batchTimeout,
-		rl:           rl,
-		filter:       broadcastfilter.NewRuleSet([]broadcastfilter.Rule{broadcastfilter.EmptyRejectRule, broadcastfilter.AcceptRule}),
-		sendChan:     make(chan *ab.BroadcastMessage),
-		exitChan:     make(chan struct{}),
+		queueSize:     queueSize,
+		batchSize:     batchSize,
+		batchTimeout:  batchTimeout,
+		rl:            rl,
+		filter:        filters,
+		configManager: configManager,
+		sendChan:      make(chan *ab.BroadcastMessage),
+		exitChan:      make(chan struct{}),
 	}
 	return bs
 }
@@ -59,41 +64,64 @@ func (bs *broadcastServer) halt() {
 
 func (bs *broadcastServer) main() {
 	var curBatch []*ab.BroadcastMessage
-outer:
+	var timer <-chan time.Time
+
+	cutBatch := func() {
+		bs.rl.Append(curBatch, nil)
+		curBatch = nil
+		timer = nil
+	}
+
 	for {
-		timer := time.After(bs.batchTimeout)
-		for {
-			select {
-			case msg := <-bs.sendChan:
-				// The messages must be filtered a second time in case configuration has changed since the message was received
-				action, _ := bs.filter.Apply(msg)
-				switch action {
-				case broadcastfilter.Accept:
-					curBatch = append(curBatch, msg)
-					if len(curBatch) < bs.batchSize {
-						continue
-					}
+		select {
+		case msg := <-bs.sendChan:
+			// The messages must be filtered a second time in case configuration has changed since the message was received
+			action, _ := bs.filter.Apply(msg)
+			switch action {
+			case broadcastfilter.Accept:
+				curBatch = append(curBatch, msg)
+
+				if len(curBatch) >= bs.batchSize {
 					logger.Debugf("Batch size met, creating block")
-				case broadcastfilter.Forward:
-					logger.Debugf("Ignoring message because it was not accepted by a filter")
-				default:
-					// TODO add support for other cases, unreachable for now
-					logger.Fatalf("NOT IMPLEMENTED YET")
+					cutBatch()
+				} else if len(curBatch) == 1 {
+					// If this is the first request in a batch, start the batch timer
+					timer = time.After(bs.batchTimeout)
 				}
-			case <-timer:
-				if len(curBatch) == 0 {
-					continue outer
+			case broadcastfilter.Reconfigure:
+				// TODO, this is unmarshaling for a second time, we need a cleaner interface, maybe Apply returns a second arg with thing to put in the batch
+				newConfig := &ab.ConfigurationEnvelope{}
+				if err := proto.Unmarshal(msg.Data, newConfig); err != nil {
+					logger.Errorf("A change was flagged as configuration, but could not be unmarshaled: %v", err)
+					continue
 				}
-				logger.Debugf("Batch timer expired, creating block")
-			case <-bs.exitChan:
-				logger.Debugf("Exiting")
-				return
+				err := bs.configManager.Apply(newConfig)
+				if err != nil {
+					logger.Warningf("A configuration change made it through the ingress filter but could not be included in a batch: %v", err)
+					continue
+				}
+
+				logger.Debugf("Configuration change applied successfully, committing previous block and configuration block")
+				cutBatch()
+				bs.rl.Append([]*ab.BroadcastMessage{msg}, nil)
+			case broadcastfilter.Reject:
+				fallthrough
+			case broadcastfilter.Forward:
+				logger.Debugf("Ignoring message because it was not accepted by a filter")
+			default:
+				logger.Fatalf("Received an unknown rule response: %v", action)
 			}
-			break
+		case <-timer:
+			if len(curBatch) == 0 {
+				logger.Warningf("Batch timer expired with no pending requests, this might indicate a bug")
+				continue
+			}
+			logger.Debugf("Batch timer expired, creating block")
+			cutBatch()
+		case <-bs.exitChan:
+			logger.Debugf("Exiting")
+			return
 		}
-
-		bs.rl.Append(curBatch, nil)
-		curBatch = nil
 	}
 }
 
@@ -139,6 +167,8 @@ func (b *broadcaster) queueBroadcastMessages(srv ab.AtomicBroadcast_BroadcastSer
 		action, _ := b.bs.filter.Apply(msg)
 
 		switch action {
+		case broadcastfilter.Reconfigure:
+			fallthrough
 		case broadcastfilter.Accept:
 			select {
 			case b.queue <- msg:
@@ -151,8 +181,7 @@ func (b *broadcaster) queueBroadcastMessages(srv ab.AtomicBroadcast_BroadcastSer
 		case broadcastfilter.Reject:
 			err = srv.Send(&ab.BroadcastResponse{ab.Status_BAD_REQUEST})
 		default:
-			// TODO add support for other cases, unreachable for now
-			logger.Fatalf("NOT IMPLEMENTED YET")
+			logger.Fatalf("Unknown filter action :%v", action)
 		}
 
 		if err != nil {