OpenflowPlugin学习笔记1

ConfigurableOpenFlowProviderModule是OpenFlowPlugin中启动加载的入口，如下：

@Override
public java.lang.AutoCloseable createInstance() {
    pluginProvider =  new OpenflowPluginProvider();
    pluginProvider.setDataBroker(getDataBrokerDependency());
    pluginProvider.setNotificationService(getNotificationServiceDependency());
    pluginProvider.setRpcRegistry(getRpcRegistryDependency());
    pluginProvider.setSwitchConnectionProviders(getOpenflowSwitchConnectionProviderDependency());
    pluginProvider.setRole(getRole()); // 此时获得的role为缺省值NOCHANGE，该值位于AbstractConfigurableOpenFlowProviderModule中
    pluginProvider.initialization();
    return pluginProvider;
}

其中创建了一个OpenflowPluginProvider，即OpenflowPlugin功能实际提供者，其中调用的initialization方法将服务进行了初始化，如下：

public void initialization() {
    messageCountProvider = new MessageSpyCounterImpl();
    extensiOnConverterManager= new ExtensionConverterManagerImpl();
    roleManager = new OFRoleManager(OFSessionUtil.getSessionManager());

    LOG.debug("dependencies gathered..");
    registratiOnManager= new SalRegistrationManager();
    registrationManager.setDataService(dataBroker);
    registrationManager.setPublishService(notificationService);
    registrationManager.setRpcProviderRegistry(rpcRegistry);
    registrationManager.init();

    mdCOntroller= new MDController();
    mdController.setSwitchConnectionProviders(switchConnectionProviders);
    mdController.setMessageSpyCounter(messageCountProvider);
    mdController.setExtensionConverterProvider(extensionConverterManager);
    mdController.init();
    mdController.start();
}

下行通道

其中可大体分成两部分，一部分是SalRegistrationManager，另一部分则是MDController。
SalRegistrationManager可以理解为管理了控制器到交换机的下行通道。
在SalRegistrationManager中init方法，注册了SessionListener，当控制器与交换机连接会话发生变化时，会触发onSessionAdded和onSessionRemoved方法，如下：

@Override
public void onSessionAdded(final SwitchSessionKeyOF sessionKey, final SessionContext context) {
    GetFeaturesOutput features = context.getFeatures(); // 获取OFPT_FEATURES_REPLY
    BigInteger datapathId = features.getDatapathId(); // 从OFPT_FEATURES_REPLY中获取dpid
    InstanceIdentifier identifier = identifierFromDatapathId(datapathId);
    NodeRef nodeRef = new NodeRef(identifier);
    NodeId nodeId = nodeIdFromDatapathId(datapathId);
    ModelDrivenSwitchImpl ofSwitch = new ModelDrivenSwitchImpl(nodeId, identifier, context); // 创建交换机实例
    CompositeObjectRegistration registration =
            ofSwitch.register(rpcProviderRegistry); // 注册rpc调用
    context.setProviderRegistration(registration);

    LOG.debug("ModelDrivenSwitch for {} registered to MD-SAL.", datapathId);

    NotificationQueueWrapper wrappedNotification = new NotificationQueueWrapper(
            nodeAdded(ofSwitch, features, nodeRef),
            context.getFeatures().getVersion());
    context.getNotificationEnqueuer().enqueueNotification(wrappedNotification);
}

可以认为每一台交换机与控制器建立连接后，控制器都会为其创建一个ModelDrivenSwitchImpl实例，并为其注册相应的rpc，而ModelDrivenSwitchImpl则实现了多个rpc接口，继承如下：

public interface ModelDrivenSwitch
        extends
        SalGroupService,
        SalFlowService,
        SalMeterService, SalTableService, SalPortService, PacketProcessingService, NodeConfigService,
        OpendaylightGroupStatisticsService, OpendaylightMeterStatisticsService, OpendaylightFlowStatisticsService,
        OpendaylightPortStatisticsService, OpendaylightFlowTableStatisticsService, OpendaylightQueueStatisticsService,
        Identifiable>

而AbstractModelDrivenSwitch则实现ModelDrivenSwitch，如下：

public abstract class AbstractModelDrivenSwitch implements ModelDrivenSwitch

在AbstractModelDrivenSwitch中，注册了所有的rpc实现为AbstractModelDrivenSwitch，由于此类为抽象类，因此具体方法的实现将由实现类完成，即ModelDrivenSwitchImpl，如下：

 @Override
public CompositeObjectRegistration register(RpcProviderRegistry rpcProviderRegistry) {
    CompositeObjectRegistrationBuilder builder = CompositeObjectRegistration
            . builderFor(this);

    final RoutedRpcRegistration flowRegistration = rpcProviderRegistry.addRoutedRpcImplementation(SalFlowService.class, this);
    flowRegistration.registerPath(NodeContext.class, getIdentifier()); // 将rpc路由表中的数据进行更新，getIdentifier() 方法带入更新节点的path
    builder.add(flowRegistration);
    
    ...
}

上述rpc接口的实现均在ModelDrivenSwitchImpl中，以SalFlowService为例，如下：

@Override
public Future> addFlow(final AddFlowInput input) {
    LOG.debug("Calling the FlowMod RPC method on MessageDispatchService");
    // use primary connection
    SwitchConnectionDistinguisher COOKIE = null;

    OFRpcTask> task =
            OFRpcTaskFactory.createAddFlowTask(rpcTaskContext, input, COOKIE);
    ListenableFuture> result = task.submit();

    return Futures.transform(result, OFRpcFutureResultTransformFactory.createForAddFlowOutput());
}

上行通道

MDController负责控制器与交换机的信令交互，即非流表、组表消息的交互，可以理解为控制器与交换机的上行通道管理。在init方法中，交换机与控制器消息处理实现被添加到映射中，如下：

OpenflowPortsUtil.init(); // 完成协议中端口定义的映射
...
// 每个translator对应了Openflow协议中一种消息，负责将Of消息转换为MD-SAL中的各个notification
addMessageTranslator(ErrorMessage.class, OF10, new ErrorV10Translator());
addMessageTranslator(ErrorMessage.class, OF13, new ErrorTranslator());
addMessageTranslator(FlowRemovedMessage.class, OF10, new FlowRemovedTranslator());
addMessageTranslator(FlowRemovedMessage.class, OF13, new FlowRemovedTranslator());
...
// 制定了每种notification的通用pulisher
addMessagePopListener(NodeErrorNotification.class, notificationPopListener);
addMessagePopListener(BadActionErrorNotification.class, notificationPopListener);
addMessagePopListener(BadInstructionErrorNotification.class, notificationPopListener);
addMessagePopListener(BadMatchErrorNotification.class, notificationPopListener);

在init后，调用start方法创建SwitchConnectionHandlerImpl负责与处理交换机连接，start方法会启动一系列SwitchConnectionHandler，这些SwitchConnectionHandler会依次处理连接，以找到一个合适的，如下：

List> starterChain = new ArrayList<>(switchConnectionProviders.size());
for (SwitchConnectionProvider switchConnectionPrv : switchConnectionProviders) {
    switchConnectionPrv.setSwitchConnectionHandler(switchConnectionHandler);
    ListenableFuture isOnlineFuture= switchConnectionPrv.startup();
    starterChain.add(isOnlineFuture);
}

角色管理

除去SalRegistrationManager与MDController，OpenflowPluginProvider还创建了一个OFRoleManager实例，在OpenflowPluginProvider中与Role相关的方法如下：

/**
 * @param role of instance
 */
public void setRole(OfpRole role) {
    this.role = role;
}

/**
 * @param newRole
 */
public void fireRoleChange(OfpRole newRole) {
    if (!role.equals(newRole)) {
        LOG.debug("my role was chaged from {} to {}", role, newRole);
        role = newRole;
        switch (role) {
        case BECOMEMASTER:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        case BECOMESLAVE:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        case NOCHANGE:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        default:
            LOG.warn("role not supported: {}", role);
            break;
        }
    }
}

从方法实现的角度看，似乎role在OpenflowPluginProvider中是一个集中控制的对象，并非与交换机节点绑定，即对于一定区域内的所有交换机而言，只能出现一个一个master，而不能出现某个交换机在某个控制器上为master的情况。从OFRoleManager提供的方法看似乎也验证了这一点，如下：

/**
 * @param sessionManager
 */
public OFRoleManager(final SessionManager sessionManager) {
    Preconditions.checkNotNull("Session manager can not be empty.", sessionManager);
    this.sessiOnManager= sessionManager;
    workQueue = new PriorityBlockingQueue<>(500, new Comparator() { // 队列容量为500
        @Override
        public int compare(final RolePushTask o1, final RolePushTask o2) {
            return Integer.compare(o1.getPriority(), o2.getPriority()); // 按优先级排列的队列
        }
    });
    ThreadPoolLoggingExecutor delegate = new ThreadPoolLoggingExecutor(
            1, 1, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue(1), "ofRoleBroadcast");
    broadcastPool = MoreExecutors.listeningDecorator(
            delegate);
}

/**
 * change role on each connected device
 *
 * @param role
 */
public void manageRoleChange(final OfpRole role) {
    for (final SessionContext session : sessionManager.getAllSessions()) { // 遍历所有的连接会话
        try {
            workQueue.put(new RolePushTask(role, session));
        } catch (InterruptedException e) {
            LOG.warn("Processing of role request failed while enqueueing role task: {}", e.getMessage());
        }
    }

    while (!workQueue.isEmpty()) {
        RolePushTask task = workQueue.poll();
        ListenableFuture rolePushResult = broadcastPool.submit(task); // 该方法会调用RolePushTask中的call方法
        CheckedFuture rolePushResultChecked =
                RoleUtil.makeCheckedRuleRequestFxResult(rolePushResult);
        try {
            Boolean succeeded = rolePushResultChecked.checkedGet(TIMEOUT, TIMEOUT_UNIT);
            if (!MoreObjects.firstNonNull(succeeded, Boolean.FALSE)) {
                if (task.getRetryCounter() 

generatiOnId= RoleUtil.getNextGenerationId(generationId);

// try to possess role on device
Future> roleReply = RoleUtil.sendRoleChangeRequest(session, role, generationId);
// flush election result with barrier
BarrierInput barrierInput = MessageFactory.createBarrier(
        session.getFeatures().getVersion(), session.getNextXid());
Future> barrierResult = session.getPrimaryConductor().getConnectionAdapter().barrier(barrierInput);

public void fireRoleChange(OfpRole newRole) {
    if (!role.equals(newRole)) {
        LOG.debug("my role was chaged from {} to {}", role, newRole);
        role = newRole;
        switch (role) {
        case BECOMEMASTER:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        case BECOMESLAVE:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        case NOCHANGE:
            //TODO: implement appropriate action
            roleManager.manageRoleChange(role);
            break;
        default:
            LOG.warn("role not supported: {}", role);
            break;
        }
    }
}

/**
 * This method is called in order to translate message to MD-SAL or from MD-SAL.
 *
 * @param COOKIE
 *            auxiliary connection identifier
 * @param sc
 *            The SessionContext which sent the OF message
 * @param msg
 *            The OF message
 *
 * @return translated message
 */
O translate(SwitchConnectionDistinguisher COOKIE, SessionContext sc, I msg);

@Override
public List translate(SwitchConnectionDistinguisher COOKIE,
        SessionContext sc, OfHeader msg) {
    if(msg instanceof OF_MESSAGE.class) { // 判断消息是否属于要处理的消息类型
        // 按消息格式填充各个字段
        ...
        return list;
    } else {
        return Collections.emptyList(); // 处理出错则返回一个空的列表
    }
}

public void start() {
    LOG.debug("starting ..");
    LOG.debug("switchConnectionProvider: " + switchConnectionProviders);
    // setup handler
    SwitchConnectionHandlerImpl switchCOnnectionHandler= new SwitchConnectionHandlerImpl(); // 实际创建了一个queuekeeper
    switchConnectionHandler.setMessageSpy(messageSpyCounter);

    errorHandler = new ErrorHandlerSimpleImpl(); // 为捕获异常创建实例

    switchConnectionHandler.setErrorHandler(errorHandler);
    switchConnectionHandler.init(); // 经过调用，真正地注册了translator和popListener

    List> starterChain = new ArrayList<>(switchConnectionProviders.size());
    for (SwitchConnectionProvider switchConnectionPrv : switchConnectionProviders) {
        switchConnectionPrv.setSwitchConnectionHandler(switchConnectionHandler);
        ListenableFuture isOnlineFuture= switchConnectionPrv.startup();
        starterChain.add(isOnlineFuture);
    }

    Future> srvStarted = Futures.allAsList(starterChain);
}