Matrix/workloadGenerator.go

package main

import (
	"bytes"
	"crypto/tls"
	"encoding/base64"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"git.nutfactory.org/hoernschen/Matrix/entities/event"
	"git.nutfactory.org/hoernschen/Matrix/entities/general"
	"git.nutfactory.org/hoernschen/Matrix/entities/room"
	"git.nutfactory.org/hoernschen/Matrix/entities/user"
	"git.nutfactory.org/hoernschen/Matrix/utils"
)

var BaseLineTest = true

var systemParamsIndex = 0

type SystemParams struct {
	Id                    string
	BytesToSend           int
	MessagesPerSecond     float32
	Distribution          map[string][]string
	Packetloss            int
	MinutesNotAvailable   int
	Consensus             bool
	AuthentificationCheck bool
	Signing               bool
	Encryption            bool
}

var httpString string
var users = []map[string][]string{
	map[string][]string{
		"143.93.38.208": []string{
			"user1",
			"user2",
			"user3",
			"user4",
			"user5",
			"user6",
		},
	},
	map[string][]string{
		"143.93.38.207": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.208": []string{
			"user1",
		},
		"143.93.38.209": []string{
			"user1",
		},
	},
	map[string][]string{
		"143.93.38.207": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.208": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
		"143.93.38.209": []string{
			"user1",
			"user2",
			"user3",
			"user4",
		},
	},
}

var servers = []string{
	"143.93.38.207",
	"143.93.38.208",
	"143.93.38.209",
}

var systemParams = []SystemParams{
	SystemParams{
		Id:                    "111111111",
		BytesToSend:           280,
		MessagesPerSecond:     1.0,
		Distribution:          users[1],
		Packetloss:            1,
		MinutesNotAvailable:   0,
		Consensus:             true,
		AuthentificationCheck: true,
		Signing:               true,
		Encryption:            true,
	},
	/*
		SystemParams{
			Id:                    "011111111",
			BytesToSend:           8,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "211111111",
			BytesToSend:           512,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "101111111",
			BytesToSend:           280,
			MessagesPerSecond:     0.1,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "121111111",
			BytesToSend:           280,
			MessagesPerSecond:     10.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "110111111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[0],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "112111111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[2],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111011111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            0,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111211111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            20,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111121111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   1,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111110111",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             false,
			AuthentificationCheck: true,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111111011",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: false,
			Signing:               true,
			Encryption:            true,
		},
		SystemParams{
			Id:                    "111111101",
			BytesToSend:           280,
			MessagesPerSecond:     1.0,
			Distribution:          users[1],
			Packetloss:            1,
			MinutesNotAvailable:   0,
			Consensus:             true,
			AuthentificationCheck: true,
			Signing:               false,
			Encryption:            true,
		},
	*/
}

var userIds []string
var accessTokens map[string]string
var roomId string

func getCSVWriter(filename string) (writer *csv.Writer) {
	file, err := os.Create(fmt.Sprintf("%s %s.csv", strconv.FormatInt(time.Now().Unix(), 10), filename))
	if err != nil {
		log.Fatalf("Error Creating CSV: %s", err)
	}
	defer file.Close()
	writer = csv.NewWriter(file)
	return
}

func main() {
	http.DefaultTransport.(*http.Transport).TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	httpString = "https"
	iteration := 0
	if BaseLineTest {
		file, err := os.Create(fmt.Sprintf("%s Baseline Measurement.csv", strconv.FormatInt(time.Now().Unix(), 10)))
		if err != nil {
			log.Fatalf("Error Creating CSV: %s", err)
		}
		defer file.Close()
		writer := csv.NewWriter(file)
		defer writer.Flush()
		err = writer.Write([]string{"Iteration", "Start", "End"})
		if err != nil {
			log.Fatalf("Error in Writing CSV: %s", err)
		}
		for iteration < 10 {
			log.Printf("Iteration: %s", strconv.Itoa(iteration))
			start := time.Now().Unix()
			time.Sleep(2 * time.Minute)
			end := time.Now().Unix()
			iteration++
			err = writer.Write([]string{strconv.Itoa(iteration), strconv.FormatInt(start, 10), strconv.FormatInt(end, 10)})
			if err != nil {
				log.Fatalf("Error in Writing CSV: %s", err)
			}
		}
	} else {
		for _, systemParam := range systemParams {
			file, err := os.Create(fmt.Sprintf("%s %s Measurement Matrix.csv", strconv.FormatInt(time.Now().Unix(), 10), systemParam.Id))
			if err != nil {
				log.Fatalf("Error Creating CSV: %s", err)
			}
			writer := csv.NewWriter(file)
			err = writer.Write([]string{"Iteration", "Start", "End", "Actions Send"})
			if err != nil {
				log.Fatalf("Error in Writing CSV: %s", err)
			}
			millisecondsToWait := 1000 / systemParam.MessagesPerSecond
			iteration = 0
			for iteration < 30 {
				err := setUp(systemParam)
				if err != nil {
					log.Fatalf("Error in SetUp: %s", err)
				}
				b := make([]byte, systemParam.BytesToSend)
				_, err = rand.Read(b)
				message := base64.RawStdEncoding.EncodeToString(b)
				start := time.Now()
				end := start.Add(2 * time.Minute).Unix()
				log.Printf("Id: %s - Iteration: %s - Start: %s - End: %s", systemParam.Id, strconv.Itoa(iteration), strconv.FormatInt(start.Unix(), 10), strconv.FormatInt(end, 10))
				actionsSend := 0
				for time.Now().Unix() < end {
					time.Sleep(time.Duration(millisecondsToWait) * time.Millisecond)

					err = sendMessage(message)
					if err != nil {
						log.Fatalf("Error sending Message %s", err)
					}

					actionsSend++
				}
				iteration++
				err = writer.Write([]string{strconv.Itoa(iteration), strconv.FormatInt(start.Unix(), 10), strconv.FormatInt(end, 10), strconv.Itoa(actionsSend)})
				if err != nil {
					log.Fatalf("Error in Writing CSV: %s", err)
				}
				time.Sleep(1 * time.Second)
				err = reset()
				if err != nil {
					log.Fatalf("Error in Reset: %s", err)
				}
				time.Sleep(1 * time.Second)
			}
			writer.Flush()
			file.Close()
		}
	}
}

func sendMessage(message string) (err error) {
	userIndex := rand.Intn(3)
	userId := userIds[userIndex]
	accessToken := accessTokens[userId]
	err, txnId := utils.CreateUUID()
	if err != nil {
		return
	}
	requestUrl := fmt.Sprintf("%s://%s/_matrix/client/r0/rooms/%s/send/m.room.message/%s", httpString, strings.Split(userId, ":")[1], roomId, txnId)
	request := event.SendMessageRequest{
		MessageType: "m.text",
		Body:        message,
	}
	reqBody, err := json.Marshal(request)
	if err != nil {
		return
	}
	client := &http.Client{}
	var req *http.Request
	req, err = http.NewRequest(http.MethodPut, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{"application/json"}
	req.Header["Authorization"] = []string{fmt.Sprintf("Bearer %s", accessToken)}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		handleError(res)
	}
	return
}

func setParams(systemParamsToUse SystemParams) (err error) {
	serverNotAvailableIndex := 1
	for i, server := range servers {
		minutesNotAvailable := 0
		if serverNotAvailableIndex == i {
			minutesNotAvailable = systemParamsToUse.MinutesNotAvailable
		}
		requestUrl := fmt.Sprintf("%s://%s/setparams", httpString, server)
		request := general.SetParamBody{
			Packetloss:            systemParamsToUse.Packetloss,
			UnavailableTill:       time.Now().Add(time.Duration(minutesNotAvailable) * time.Minute).Unix(),
			Consensus:             systemParamsToUse.Consensus,
			AuthentificationCheck: systemParamsToUse.AuthentificationCheck,
			Signing:               systemParamsToUse.Signing,
			Encryption:            systemParamsToUse.Encryption,
		}
		var reqBody []byte
		reqBody, err = json.Marshal(request)
		if err != nil {
			return
		}
		client := &http.Client{Timeout: 10 * time.Second}
		var req *http.Request
		req, err = http.NewRequest(http.MethodGet, requestUrl, bytes.NewBuffer(reqBody))
		if err != nil {
			return
		}
		req.Header["Content-Type"] = []string{"application/json"}
		var res *http.Response
		res, err = client.Do(req)
		if err != nil {
			return
		}
		if res.StatusCode != http.StatusOK {
			handleError(res)
		}
	}
	return
}

func setUp(systemParamsToUse SystemParams) (err error) {
	accessTokens = make(map[string]string)
	err = createUsers(systemParamsToUse.Distribution)
	if err != nil {
		log.Printf("Error in User-Creation: %s", err)
		return
	}
	// Create Room
	err = createRoom(userIds[0])
	if err != nil {
		log.Printf("Error in Room-Creation: %s", err)
		return
	}

	// Join to Room
	err = joinRoom(userIds[1:])
	if err != nil {
		log.Printf("Error joining Room: %s", err)
		return
	}

	err = setParams(systemParamsToUse)
	return
}

func reset() (err error) {
	roomId = ""
	userIds = []string{}
	accessTokens = make(map[string]string)
	for _, server := range servers {
		requestUrl := fmt.Sprintf("%s://%s/reset", httpString, server)
		client := &http.Client{Timeout: 10 * time.Second}
		var req *http.Request
		req, err = http.NewRequest(http.MethodGet, requestUrl, bytes.NewBuffer(nil))
		if err != nil {
			return
		}
		req.Header["Content-Type"] = []string{"application/json"}
		var res *http.Response
		res, err = client.Do(req)
		if err != nil {
			return
		}
		if res.StatusCode != http.StatusOK {
			handleError(res)
		}
	}
	return
}

func createRoom(userId string) (err error) {
	accessToken := accessTokens[userId]
	requestUrl := fmt.Sprintf("%s://%s/_matrix/client/r0/createRoom", httpString, strings.Split(userId, ":")[1])
	request := room.CreateRoomRequest{
		Visibility:  "public",
		Name:        "Testraum",
		Topic:       "Raum für die Energieeffizienz-Tests",
		RoomVersion: "1",
		CreationContent: room.CreationContent{
			Federated: true,
		},
		Preset:   "public_chat",
		IsDirect: true,
	}
	reqBody, err := json.Marshal(request)
	if err != nil {
		return
	}
	client := &http.Client{Timeout: 10 * time.Second}
	req, err := http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{"application/json"}
	req.Header["Authorization"] = []string{fmt.Sprintf("Bearer %s", accessToken)}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		handleError(res)
	} else {
		response := room.CreateRoomResponse{}
		decoder := json.NewDecoder(res.Body)
		err = decoder.Decode(&response)
		if err != nil {
			return
		}
		roomId = response.RoomId
		log.Printf("Room created: %s", roomId)
	}
	return
}

func joinRoom(userIdsToJoin []string) (err error) {
	for _, userId := range userIdsToJoin {
		err = joinRoomUser(userId)
		if err != nil {
			return
		}
		//time.Sleep(time.Duration(1) * time.Second)
	}
	return
}

func joinRoomUser(userId string) (err error) {
	log.Printf("Join %s to %s", userId, roomId)
	accessToken := accessTokens[userId]
	requestUrl := fmt.Sprintf("%s://%s/_matrix/client/r0/rooms/%s/join", httpString, strings.Split(userId, ":")[1], roomId)
	request := room.JoinRoomUserRequest{}
	reqBody, err := json.Marshal(request)
	if err != nil {
		return
	}
	client := &http.Client{Timeout: 10 * time.Second}
	req, err := http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{"application/json"}
	req.Header["Authorization"] = []string{fmt.Sprintf("Bearer %s", accessToken)}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		handleError(res)
	} else {
		response := room.JoinRoomUserResponse{}
		decoder := json.NewDecoder(res.Body)
		err = decoder.Decode(&response)
		if err != nil {
			return
		}
		roomId = response.RoomId
		log.Printf("%s joined Room", userId)
	}
	return
}

func createUsers(serverUserMap map[string][]string) (err error) {
	for server, usersToCreate := range serverUserMap {
		for _, userToCreate := range usersToCreate {
			var userId string
			var accessToken string
			userId, accessToken, err = createUser(userToCreate, server)
			if err != nil {
				return
			}
			if userId != "" && accessToken != "" {
				log.Printf("%s created - AccessToken: %s", userId, accessToken)
				accessTokens[userId] = accessToken
				userIds = append(userIds, userId)
			}
		}
	}
	return
}

func createUser(userToCreate string, homeserver string) (userId string, accessToken string, err error) {
	requestUrl := fmt.Sprintf("%s://%s/_matrix/client/r0/register", httpString, homeserver)
	request := user.RegisterRequest{
		Auth: user.AuthentificationData{
			LoginType: "m.login.password",
		},
		Username:   userToCreate,
		Password:   "password",
		DeviceName: fmt.Sprintf("%s's device", userToCreate),
	}
	reqBody, err := json.Marshal(request)
	if err != nil {
		return
	}
	client := &http.Client{Timeout: 10 * time.Second}
	req, err := http.NewRequest(http.MethodPost, requestUrl, bytes.NewBuffer(reqBody))
	if err != nil {
		return
	}
	req.Header["Content-Type"] = []string{"application/json"}
	res, err := client.Do(req)
	if err != nil {
		return
	}
	if res.StatusCode != http.StatusOK {
		handleError(res)
	} else {
		response := user.RegisterResponse{}
		decoder := json.NewDecoder(res.Body)
		err = decoder.Decode(&response)
		if err != nil {
			return
		}
		userId = response.UserId
		accessToken = response.AccessToken
	}
	return
}

func handleError(res *http.Response) {
	utils.HandleHTTPError(res)
}