<commit_before>//
// Copyright (c) 2016 Intel Corporation
//
// 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 main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"runtime/pprof"
	"sync"
	"syscall"
	"time"

	"github.com/01org/ciao/clogger/gloginterface"
	"github.com/01org/ciao/osprepare"
	"github.com/01org/ciao/payloads"
	"github.com/01org/ciao/ssntp"
	"github.com/golang/glog"
	"github.com/pkg/errors"
	"gopkg.in/yaml.v2"
)

var cert = flag.String("cert", "/etc/pki/ciao/cert-Scheduler-localhost.pem", "Server certificate")
var cacert = flag.String("cacert", "/etc/pki/ciao/CAcert-server-localhost.pem", "CA certificate")
var cpuprofile = flag.String("cpuprofile", "", "Write cpu profile to file")
var heartbeat = flag.Bool("heartbeat", false, "Emit status heartbeat text")
var logDir = "/var/lib/ciao/logs/scheduler"
var configURI = flag.String("configuration-uri", "file:///etc/ciao/configuration.yaml",
	"Cluster configuration URI")

type ssntpSchedulerServer struct {
	// user config overrides ------------------------------------------
	heartbeat  bool
	cpuprofile string

	// ssntp ----------------------------------------------------------
	config *ssntp.Config
	ssntp  ssntp.Server

	// scheduler internal state ---------------------------------------

	// Command & Status Reporting node(s)
	controllerMap   map[string]*controllerStat
	controllerList  []*controllerStat // 1 controllerMaster at front of list
	controllerMutex sync.RWMutex      // Rlock traversing map, Lock modifying map

	// Compute Nodes
	cnMap      map[string]*nodeStat
	cnList     []*nodeStat
	cnMutex    sync.RWMutex // Rlock traversing map, Lock modifying map
	cnMRU      *nodeStat
	cnMRUIndex int
	//cnInactiveMap      map[string]nodeStat

	// Network Nodes
	nnMap      map[string]*nodeStat
	nnList     []*nodeStat
	nnMutex    sync.RWMutex // Rlock traversing map, Lock modifying map
	nnMRU      *nodeStat
	nnMRUIndex int
}

func newSsntpSchedulerServer() *ssntpSchedulerServer {
	return &ssntpSchedulerServer{
		controllerMap: make(map[string]*controllerStat),
		cnMap:         make(map[string]*nodeStat),
		cnMRUIndex:    -1,
		nnMap:         make(map[string]*nodeStat),
		nnMRUIndex:    -1,
	}
}

type nodeStat struct {
	mutex       sync.Mutex
	status      ssntp.Status
	uuid        string
	memTotalMB  int
	memAvailMB  int
	diskTotalMB int
	diskAvailMB int
	load        int
	cpus        int
	isNetNode   bool
	networks    []payloads.NetworkStat
}

type controllerStatus uint8

func (s controllerStatus) String() string {
	switch s {
	case controllerMaster:
		return "MASTER"
	case controllerBackup:
		return "BACKUP"
	}

	return ""
}

const (
	controllerMaster controllerStatus = iota
	controllerBackup
)

type controllerStat struct {
	mutex  sync.Mutex
	status controllerStatus
	uuid   string
}

func prepareNodeConnectionEvent(nodeUUID string, nodeType payloads.Resource, connected bool) (b []byte, err error) {
	event := payloads.NodeConnectedEvent{
		NodeUUID: nodeUUID,
		NodeType: nodeType,
	}

	if connected == true {
		payload := payloads.NodeConnected{
			Connected: event,
		}

		b, err = yaml.Marshal(&payload)
	} else {
		payload := payloads.NodeDisconnected{
			Disconnected: event,
		}

		b, err = yaml.Marshal(&payload)
	}

	return
}

// The ssntp server implementation is expected to generate ssntp client
// connect/disconnect events. This function sends to one controller a
// connect event for each currently connected nodes.
func (sched *ssntpSchedulerServer) sendDirectedNodeConnectionEvents(ctlUUID string) {
	sched.cnMutex.RLock()
	defer sched.cnMutex.RUnlock()
	for _, node := range sched.cnList {
		nodeUUID := node.uuid

		b, err := prepareNodeConnectionEvent(nodeUUID, payloads.ComputeNode, true)
		if err != nil {
			errors.Wrap(err, "Node connection event lost")
			continue
		}

		sched.ssntp.SendEvent(ctlUUID, ssntp.NodeConnected, b)
	}

	sched.nnMutex.RLock()
	defer sched.nnMutex.RUnlock()
	for _, node := range sched.nnList {
		nodeUUID := node.uuid

		b, err := prepareNodeConnectionEvent(nodeUUID, payloads.NetworkNode, true)
		if err != nil {
			errors.Wrap(err, "Node connection event lost")
			continue
		}

		sched.ssntp.SendEvent(ctlUUID, ssntp.NodeConnected, b)
	}
}

// The ssntp server implementation is expected to generate ssntp client
// connect/disconnect events. This function sends them to all controllers.
func (sched *ssntpSchedulerServer) sendNodeConnectionEvents(nodeUUID string, nodeType payloads.Resource, connected bool) {
	b, err := prepareNodeConnectionEvent(nodeUUID, nodeType, connected)
	if err != nil {
		errors.Wrap(err, "Node connection event lost")
	}

	sched.controllerMutex.RLock()
	defer sched.controllerMutex.RUnlock()

	for _, ctl := range sched.controllerMap {
		sched.ssntp.SendEvent(ctl.uuid, ssntp.NodeConnected, b)
	}
}

// Add state for newly connected Controller
// This function is symmetric with disconnectController().<commit_msg>ciao-scheduler: Send node disconnect messaged with correct type<commit_after>//
// Copyright (c) 2016 Intel Corporation
//
// 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 main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"runtime/pprof"
	"sync"
	"syscall"
	"time"

	"github.com/01org/ciao/clogger/gloginterface"
	"github.com/01org/ciao/osprepare"
	"github.com/01org/ciao/payloads"
	"github.com/01org/ciao/ssntp"
	"github.com/golang/glog"
	"github.com/pkg/errors"
	"gopkg.in/yaml.v2"
)

var cert = flag.String("cert", "/etc/pki/ciao/cert-Scheduler-localhost.pem", "Server certificate")
var cacert = flag.String("cacert", "/etc/pki/ciao/CAcert-server-localhost.pem", "CA certificate")
var cpuprofile = flag.String("cpuprofile", "", "Write cpu profile to file")
var heartbeat = flag.Bool("heartbeat", false, "Emit status heartbeat text")
var logDir = "/var/lib/ciao/logs/scheduler"
var configURI = flag.String("configuration-uri", "file:///etc/ciao/configuration.yaml",
	"Cluster configuration URI")

type ssntpSchedulerServer struct {
	// user config overrides ------------------------------------------
	heartbeat  bool
	cpuprofile string

	// ssntp ----------------------------------------------------------
	config *ssntp.Config
	ssntp  ssntp.Server

	// scheduler internal state ---------------------------------------

	// Command & Status Reporting node(s)
	controllerMap   map[string]*controllerStat
	controllerList  []*controllerStat // 1 controllerMaster at front of list
	controllerMutex sync.RWMutex      // Rlock traversing map, Lock modifying map

	// Compute Nodes
	cnMap      map[string]*nodeStat
	cnList     []*nodeStat
	cnMutex    sync.RWMutex // Rlock traversing map, Lock modifying map
	cnMRU      *nodeStat
	cnMRUIndex int
	//cnInactiveMap      map[string]nodeStat

	// Network Nodes
	nnMap      map[string]*nodeStat
	nnList     []*nodeStat
	nnMutex    sync.RWMutex // Rlock traversing map, Lock modifying map
	nnMRU      *nodeStat
	nnMRUIndex int
}

func newSsntpSchedulerServer() *ssntpSchedulerServer {
	return &ssntpSchedulerServer{
		controllerMap: make(map[string]*controllerStat),
		cnMap:         make(map[string]*nodeStat),
		cnMRUIndex:    -1,
		nnMap:         make(map[string]*nodeStat),
		nnMRUIndex:    -1,
	}
}

type nodeStat struct {
	mutex       sync.Mutex
	status      ssntp.Status
	uuid        string
	memTotalMB  int
	memAvailMB  int
	diskTotalMB int
	diskAvailMB int
	load        int
	cpus        int
	isNetNode   bool
	networks    []payloads.NetworkStat
}

type controllerStatus uint8

func (s controllerStatus) String() string {
	switch s {
	case controllerMaster:
		return "MASTER"
	case controllerBackup:
		return "BACKUP"
	}

	return ""
}

const (
	controllerMaster controllerStatus = iota
	controllerBackup
)

type controllerStat struct {
	mutex  sync.Mutex
	status controllerStatus
	uuid   string
}

func prepareNodeConnectionEvent(nodeUUID string, nodeType payloads.Resource, connected bool) (b []byte, err error) {
	event := payloads.NodeConnectedEvent{
		NodeUUID: nodeUUID,
		NodeType: nodeType,
	}

	if connected == true {
		payload := payloads.NodeConnected{
			Connected: event,
		}

		b, err = yaml.Marshal(&payload)
	} else {
		payload := payloads.NodeDisconnected{
			Disconnected: event,
		}

		b, err = yaml.Marshal(&payload)
	}

	return
}

// The ssntp server implementation is expected to generate ssntp client
// connect/disconnect events. This function sends to one controller a
// connect event for each currently connected nodes.
func (sched *ssntpSchedulerServer) sendDirectedNodeConnectionEvents(ctlUUID string) {
	sched.cnMutex.RLock()
	defer sched.cnMutex.RUnlock()
	for _, node := range sched.cnList {
		nodeUUID := node.uuid

		b, err := prepareNodeConnectionEvent(nodeUUID, payloads.ComputeNode, true)
		if err != nil {
			errors.Wrap(err, "Node connection event lost")
			continue
		}

		sched.ssntp.SendEvent(ctlUUID, ssntp.NodeConnected, b)
	}

	sched.nnMutex.RLock()
	defer sched.nnMutex.RUnlock()
	for _, node := range sched.nnList {
		nodeUUID := node.uuid

		b, err := prepareNodeConnectionEvent(nodeUUID, payloads.NetworkNode, true)
		if err != nil {
			errors.Wrap(err, "Node connection event lost")
			continue
		}

		sched.ssntp.SendEvent(ctlUUID, ssntp.NodeConnected, b)
	}
}

// The ssntp server implementation is expected to generate ssntp client
// connect/disconnect events. This function sends them to all controllers.
func (sched *ssntpSchedulerServer) sendNodeConnectionEvents(nodeUUID string, nodeType payloads.Resource, connected bool) {
	b, err := prepareNodeConnectionEvent(nodeUUID, nodeType, connected)
	if err != nil {
		errors.Wrap(err, "Node connection event lost")
	}

	sched.controllerMutex.RLock()
	defer sched.controllerMutex.RUnlock()

	for _, ctl := range sched.controllerMap {
		if connected {
			sched.ssntp.SendEvent(ctl.uuid, ssntp.NodeConnected, b)
		} else {
			sched.ssntp.SendEvent(ctl.uuid, ssntp.NodeDisconnected, b)
		}
	}
}

// Add state for newly connected Controller
// This function is symmetric with disconnectController().
{'fraction_non_alphanumeric': 0.1259212082613047, 'fraction_numerical': 0.0038213083431898826, 'mean_word_length': 3.452004860267315, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6510054', 'n_tokens_mistral': 3581, 'n_tokens_neox': 3477, 'n_words': 960}