Golang store.Manager类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Golang中github.com/sburnett/transformer/store.Manager类的典型用法代码示例。如果您正苦于以下问题：Golang Manager类的具体用法？Golang Manager怎么用？Golang Manager使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了Manager类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Golang代码示例。

示例1: makeSummaryByTimestampCsvWriter

func makeSummaryByTimestampCsvWriter(manager store.Manager, name string) store.Writer {
	keyNames := []string{
		"experiment",
		"node",
		"timestamp",
	}
	valueNames := []string{
		"count",
	}
	for _, i := range []int{0, 1, 5, 10, 25, 50, 75, 90, 95, 99, 100} {
		valueNames = append(valueNames, humanize.Ordinal(i))
	}
	arguments := []interface{}{
		name,
		keyNames,
		valueNames,
		new(string), // experiment
		new(string), // node
		new(int64),  // timestamp
		new(int64),  // count
	}
	for _ = range []int{0, 1, 5, 10, 25, 50, 75, 90, 95, 99, 100} {
		arguments = append(arguments, new(int64))
	}
	return manager.Writer(arguments...)
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:26，代码来源:summary.go

示例2: PackagesPipeline

func PackagesPipeline(levelDbManager, csvManager, sqliteManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	installedPackagesStore := levelDbManager.ReadingWriter("installed-packages")
	versionChangesStore := levelDbManager.ReadingWriter("version-changes")
	var node, packageName string
	var timestamp int64
	var version string
	csvStore := csvManager.Writer("packages.csv", []string{"node", "package", "timestamp"}, []string{"version"}, &node, &packageName, &timestamp, &version)
	sqliteStore := sqliteManager.Writer("packages", []string{"node", "package", "timestamp"}, []string{"version"}, &node, &packageName, &timestamp, &version)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "OpkgListInstalled",
			Reader:      ReadOnlySomeLogs(logsStore, "opkg_list-installed"),
			Transformer: transformer.MakeDoFunc(extractInstalledPackages),
			Writer:      installedPackagesStore,
		},
		transformer.PipelineStage{
			Name:        "DetectVersionChanges",
			Reader:      installedPackagesStore,
			Transformer: transformer.TransformFunc(detectChangedPackageVersions),
			Writer:      versionChangesStore,
		},
		transformer.PipelineStage{
			Name:   "WriteVersionChangesSqlite",
			Reader: versionChangesStore,
			Writer: sqliteStore,
		},
		transformer.PipelineStage{
			Name:   "WriteVersionChangesCsv",
			Reader: versionChangesStore,
			Writer: csvStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:34，代码来源:packages.go

示例3: FilterNodesPipeline

func FilterNodesPipeline(nodeId string, levelDbManager store.Manager) transformer.Pipeline {
	tracesStore := levelDbManager.Seeker("traces")
	filteredStore := levelDbManager.Writer(fmt.Sprintf("filtered-%s", nodeId))
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:   "FilterNode",
			Reader: FilterNodes(tracesStore, nodeId),
			Writer: filteredStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:11，代码来源:filtertraces.go

示例4: CsvPipeline

func CsvPipeline(levelDbManager, csvManager store.Manager) transformer.Pipeline {
	var experiment, node, filename string
	var receivedTimestamp, creationTimestamp, size int64
	csvStore := csvManager.Writer("stats.csv", []string{"experiment", "node", "filename"}, []string{"received_timestamp", "creation_timestamp", "size"}, &experiment, &node, &filename, &receivedTimestamp, &creationTimestamp, &size)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:   "WriteStatsCsv",
			Reader: levelDbManager.Reader("stats"),
			Writer: csvStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:12，代码来源:csv.go

示例5: RebootsPipeline

func RebootsPipeline(levelDbManager, csvManager, sqliteManager store.Manager) transformer.Pipeline {
	uptimeStore := levelDbManager.Seeker("uptime")
	rebootsStore := levelDbManager.ReadingWriter("reboots")
	var node string
	var timestamp int64
	rebootsCsvStore := csvManager.Writer("reboots.csv", []string{"node", "boot_timestamp"}, []string{}, &node, &timestamp)
	rebootsSqliteStore := sqliteManager.Writer("reboots", []string{"node", "boot_timestamp"}, []string{}, &node, &timestamp)

	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "InferReboots",
			Reader:      uptimeStore,
			Transformer: transformer.TransformFunc(inferReboots),
			Writer:      rebootsStore,
		},
		transformer.PipelineStage{
			Name:   "WriteRebootsCsv",
			Reader: rebootsStore,
			Writer: rebootsCsvStore,
		},
		transformer.PipelineStage{
			Name:   "WriteRebootsSqlite",
			Reader: rebootsStore,
			Writer: rebootsSqliteStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:27，代码来源:reboots.go

示例6: UptimePipeline

func UptimePipeline(levelDbManager, csvManager, sqliteManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	uptimeStore := levelDbManager.ReadingWriter("uptime")
	var node string
	var timestamp, uptime int64
	csvStore := csvManager.Writer("uptime.csv", []string{"node", "timestamp"}, []string{"uptime"}, &node, &timestamp, &uptime)
	sqliteStore := sqliteManager.Writer("uptime", []string{"node", "timestamp"}, []string{"uptime"}, &node, &timestamp, &uptime)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "Uptime",
			Reader:      ReadOnlySomeLogs(logsStore, "uptime"),
			Transformer: transformer.MakeMapFunc(extractUptime),
			Writer:      uptimeStore,
		},
		transformer.PipelineStage{
			Name:   "WriteUptimeCsv",
			Reader: uptimeStore,
			Writer: csvStore,
		},
		transformer.PipelineStage{
			Name:   "WriteUptimeSqlite",
			Reader: uptimeStore,
			Writer: sqliteStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:26，代码来源:uptime.go

示例7: MemoryUsagePipeline

func MemoryUsagePipeline(levelDbManager, csvManager, sqliteManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	memoryUsageStore := levelDbManager.ReadingWriter("memory")
	var node string
	var timestamp, used, free int64
	csvStore := csvManager.Writer("memory.csv", []string{"node", "timestamp"}, []string{"used", "free"}, &node, &timestamp, &used, &free)
	sqliteStore := sqliteManager.Writer("memory", []string{"node", "timestamp"}, []string{"used", "free"}, &node, &timestamp, &used, &free)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "Memory",
			Reader:      ReadOnlySomeLogs(logsStore, "top"),
			Transformer: transformer.MakeDoFunc(extractMemoryUsage),
			Writer:      memoryUsageStore,
		},
		transformer.PipelineStage{
			Name:   "WriteMemoryUsageCsv",
			Reader: memoryUsageStore,
			Writer: csvStore,
		},
		transformer.PipelineStage{
			Name:   "WriteMemoryUsageSqlite",
			Reader: memoryUsageStore,
			Writer: sqliteStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:26，代码来源:memory.go

示例8: DevicesCountPipeline

func DevicesCountPipeline(levelDbManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	devicesCountStore := levelDbManager.Writer("devices-count")
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "ExtractEthernetCount",
			Reader:      ReadOnlySomeLogs(logsStore, "swconfig_ports"),
			Transformer: transformer.MakeDoFunc(extractEthernetCount),
			Writer:      devicesCountStore,
		},
		transformer.PipelineStage{
			Name:        "ExtractWirelessCount",
			Reader:      ReadOnlySomeLogs(logsStore, "iw_station_count"),
			Transformer: transformer.MakeDoFunc(extractWirelessCount),
			Writer:      devicesCountStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:18，代码来源:devicescount.go

示例9: TimesCsvPipeline

func TimesCsvPipeline(levelDbManager store.Manager, csvRoot string) transformer.Pipeline {
	writeTimesCsv := func(inputChan, outputChan chan *store.Record) {
		var currentHandle *os.File
		var currentExperiment, currentNode string
		for record := range inputChan {
			var statsKey StatsKey
			lex.DecodeOrDie(record.Key, &statsKey)
			var statsValue StatsValue
			lex.DecodeOrDie(record.Value, &statsValue)

			if currentExperiment != statsKey.Experiment || currentNode != statsKey.Node {
				if currentHandle != nil {
					currentHandle.Close()
				}
				currentExperiment = statsKey.Experiment
				currentNode = statsKey.Node

				csvName := fmt.Sprintf("%s_%s.csv", currentExperiment, currentNode)
				newHandle, err := os.Create(filepath.Join(csvRoot, csvName))
				if err != nil {
					panic(err)
				}
				currentHandle = newHandle
			}

			if _, err := fmt.Fprintf(currentHandle, "%d,%d\n", statsValue.CreationTimestamp, statsValue.ReceivedTimestamp); err != nil {
				panic(err)
			}
		}
		if currentHandle != nil {
			currentHandle.Close()
		}
	}

	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "WriteTimesCsv",
			Reader:      levelDbManager.Reader("stats"),
			Transformer: transformer.TransformFunc(writeTimesCsv),
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:42，代码来源:timescsv.go

示例10: runFilterSessionsPipeline

func runFilterSessionsPipeline(startSecs, endSecs int64, levelDbManager store.Manager) {

	transformer.RunPipeline(FilterSessionsPipeline(startSecs, endSecs, levelDbManager, "test"))

	filteredStore := levelDbManager.Reader("test")
	filteredStore.BeginReading()
	for {
		record, err := filteredStore.ReadRecord()
		if err != nil {
			panic(err)
		}
		if record == nil {
			break
		}
		var traceKey TraceKey
		lex.DecodeOrDie(record.Key, &traceKey)
		fmt.Printf("%s %d %d\n", traceKey.NodeId, traceKey.SessionId, traceKey.SequenceNumber)
	}
	filteredStore.EndReading()
}

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:20，代码来源:filtertraces_test.go

示例11: IndexTarballsPipeline

func IndexTarballsPipeline(tarballsPath string, levelDbManager store.Manager) transformer.Pipeline {
	allTarballsPattern := filepath.Join(tarballsPath, "all", "health", "*", "*", "health_*.tar.gz")
	dailyTarballsPattern := filepath.Join(tarballsPath, "by-date", "*", "health", "*", "health_*.tar.gz")
	tarnamesStore := levelDbManager.ReadingWriter("tarnames")
	tarnamesIndexedStore := levelDbManager.ReadingWriter("tarnames-indexed")
	logsStore := levelDbManager.Writer("logs")
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:   "ScanLogTarballs",
			Reader: store.NewGlobReader(allTarballsPattern),
			Writer: tarnamesStore,
		},
		transformer.PipelineStage{
			Name:   "ScanDailyLogTarballs",
			Reader: store.NewGlobReader(dailyTarballsPattern),
			Writer: tarnamesStore,
		},
		transformer.PipelineStage{
			Name:        "ReadLogTarballs",
			Reader:      store.NewDemuxingReader(tarnamesStore, tarnamesIndexedStore),
			Transformer: transformer.MakeMultipleOutputsGroupDoFunc(IndexTarballs, 2),
			Writer:      store.NewMuxingWriter(logsStore, tarnamesIndexedStore),
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:25，代码来源:index.go

示例12: makeSummaryCsvWriter

func makeSummaryCsvWriter(manager store.Manager, name string) store.Writer {
	keyNames := []string{
		"experiment",
		"node",
	}
	valueNames := []string{
		"count",
	}
	for i := 0; i <= 100; i += 5 {
		valueNames = append(valueNames, humanize.Ordinal(i))
	}
	arguments := []interface{}{
		name,
		keyNames,
		valueNames,
		new(string), // experiment
		new(string), // node
		new(int64),  // count
	}
	for i := 0; i <= 100; i += 5 {
		arguments = append(arguments, new(int64))
	}
	return manager.Writer(arguments...)
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:24，代码来源:summary.go

示例13: FilterSessionsPipeline

func FilterSessionsPipeline(sessionStartTime, sessionEndTime int64, levelDbManager store.Manager, outputName string) transformer.Pipeline {
	tracesStore := levelDbManager.Reader("traces")
	traceKeyRangesStore := levelDbManager.Reader("availability-done")
	filteredStore := levelDbManager.Writer(outputName)
	parameters := filterSessions{
		SessionStartTime: sessionStartTime * 1000000,
		SessionEndTime:   sessionEndTime * 1000000,
	}
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "FilterSessions",
			Reader:      store.NewDemuxingReader(traceKeyRangesStore, tracesStore),
			Transformer: parameters,
			Writer:      filteredStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:17，代码来源:filtertraces.go

示例14: IndexTarballsPipeline

func IndexTarballsPipeline(tarballsPath string, levelDbManager store.Manager) transformer.Pipeline {
	tarballsPattern := filepath.Join(tarballsPath, "*", "*", "*.tar.gz")
	tarnamesStore := levelDbManager.ReadingWriter("tarnames")
	tarnamesIndexedStore := levelDbManager.ReadingWriter("tarnames-indexed")
	tracesStore := levelDbManager.Writer("traces")
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:   "ScanTraceTarballs",
			Reader: store.NewGlobReader(tarballsPattern),
			Writer: tarnamesStore,
		},
		transformer.PipelineStage{
			Name:        "IndexTraces",
			Transformer: transformer.MakeMultipleOutputsGroupDoFunc(IndexTarballs, 2),
			Reader:      store.NewDemuxingReader(tarnamesStore, tarnamesIndexedStore),
			Writer:      store.NewMuxingWriter(tracesStore, tarnamesIndexedStore),
		},
	}
}

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:19，代码来源:index.go

示例15: DisjointPackagesPipeline

func DisjointPackagesPipeline(levelDbManager, csvManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	disjointPackagesStore := levelDbManager.ReadingWriter("disjoint-packages")
	var filename, node string
	var timestamp int64
	csvStore := csvManager.Writer("not-disjoint.csv", []string{"filename", "node", "timestamp"}, []string{}, &filename, &node, &timestamp)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "DisjointPackages",
			Reader:      logsStore,
			Transformer: transformer.MakeDoFunc(detectDisjointPackagesError),
			Writer:      disjointPackagesStore,
		},
		transformer.PipelineStage{
			Name:   "WriteDisjointPackagesCsv",
			Reader: disjointPackagesStore,
			Writer: csvStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:20，代码来源:disjoint.go

示例16: IpRoutePipeline

func IpRoutePipeline(levelDbManager, sqliteManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	defaultRoutesStore := levelDbManager.ReadingWriter("default-routes")
	var node string
	var timestamp int64
	var gateway string
	sqliteStore := sqliteManager.Writer("defaultroutes", []string{"node", "timestamp"}, []string{"gateway"}, &node, &timestamp, &gateway)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "ExtractDefaultRoute",
			Reader:      ReadOnlySomeLogs(logsStore, "iproute"),
			Transformer: transformer.MakeDoFunc(extractDefaultRoute),
			Writer:      defaultRoutesStore,
		},
		transformer.PipelineStage{
			Name:   "WriteDefaultRoutesSqlite",
			Reader: defaultRoutesStore,
			Writer: sqliteStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:21，代码来源:iproute.go

示例17: FilesystemUsagePipeline

func FilesystemUsagePipeline(levelDbManager, csvManager store.Manager) transformer.Pipeline {
	logsStore := levelDbManager.Seeker("logs")
	filesystemUsageStore := levelDbManager.ReadingWriter("filesystem")
	var mount, node string
	var timestamp, used, free int64
	csvStore := csvManager.Writer("filesystem.csv", []string{"mount", "node", "timestamp"}, []string{"used", "free"}, &mount, &node, &timestamp, &used, &free)

	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "Filesystem",
			Reader:      ReadOnlySomeLogs(logsStore, "df"),
			Transformer: transformer.MakeDoFunc(extractFilesystemUsage),
			Writer:      filesystemUsageStore,
		},
		transformer.PipelineStage{
			Name:   "WriteFilesystemUsageCsv",
			Reader: filesystemUsageStore,
			Writer: csvStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:21，代码来源:filesystem.go

示例18: BytesPerDomainPipeline

func BytesPerDomainPipeline(levelDbManager store.Manager, bytesPerDomainPostgresStore store.Writer) transformer.Pipeline {
	tracesStore := levelDbManager.Seeker("traces")
	availabilityIntervalsStore := levelDbManager.Seeker("consistent-ranges")
	traceKeyRangesStore := levelDbManager.ReadingDeleter("bytesperdomain-trace-key-ranges")
	consolidatedTraceKeyRangesStore := levelDbManager.ReadingDeleter("bytesperdomain-consolidated-trace-key-ranges")
	addressIdTableStore := levelDbManager.SeekingWriter("bytesperdomain-address-id-table")
	aRecordTableStore := levelDbManager.SeekingWriter("bytesperdomain-a-record-table")
	cnameRecordTableStore := levelDbManager.SeekingWriter("bytesperdomain-cname-record-table")
	flowIpsTableStore := levelDbManager.SeekingWriter("bytesperdomain-flow-ips-table")
	addressIpTableStore := levelDbManager.SeekingWriter("bytesperdomain-address-ip-table")
	bytesPerTimestampShardedStore := levelDbManager.SeekingWriter("bytesperdomain-bytes-per-timestamp-sharded")
	whitelistStore := levelDbManager.SeekingWriter("bytesperdomain-whitelist")
	aRecordsWithMacStore := levelDbManager.SeekingWriter("bytesperdomain-a-records-with-mac")
	cnameRecordsWithMacStore := levelDbManager.SeekingWriter("bytesperdomain-cname-records-with-mac")
	allDnsMappingsStore := levelDbManager.SeekingWriter("bytesperdomain-all-dns-mappings")
	allWhitelistedMappingsStore := levelDbManager.SeekingWriter("bytesperdomain-all-whitelisted-mappings")
	flowMacsTableStore := levelDbManager.SeekingWriter("bytesperdomain-flow-macs-table")
	flowDomainsTableStore := levelDbManager.SeekingWriter("bytesperdomain-flow-domains-table")
	flowDomainsGroupedTableStore := levelDbManager.SeekingWriter("bytesperdomain-flow-domains-grouped-table")
	bytesPerDomainShardedStore := levelDbManager.ReadingWriter("bytesperdomain-bytes-per-domain-sharded")
	bytesPerDomainPerDeviceStore := levelDbManager.ReadingWriter("bytesperdomain-bytes-per-domain-per-device")
	bytesPerDomainStore := levelDbManager.ReadingWriter("bytesperdomain-bytes-per-domain")
	sessionsStore := levelDbManager.ReadingDeleter("bytesperdomain-sessions")
	excludeOldSessions := func(stor store.Seeker) store.Seeker {
		return store.NewPrefixIncludingReader(stor, sessionsStore)
	}
	newTracesStore := store.NewRangeExcludingReader(store.NewRangeIncludingReader(tracesStore, availabilityIntervalsStore), traceKeyRangesStore)
	return append([]transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "BytesPerDomainMapper",
			Reader:      newTracesStore,
			Transformer: transformer.MakeMultipleOutputsDoFunc(bytesPerDomainMapper, 7),
			Writer:      store.NewMuxingWriter(addressIdTableStore, aRecordTableStore, cnameRecordTableStore, flowIpsTableStore, addressIpTableStore, bytesPerTimestampShardedStore, whitelistStore),
		},
		SessionPipelineStage(newTracesStore, sessionsStore),
		transformer.PipelineStage{
			Name:        "JoinAAddressIdsWithMacAddresses",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(addressIdTableStore, aRecordTableStore)),
			Transformer: transformer.TransformFunc(joinAddressIdsWithMacAddresses),
			Writer:      aRecordsWithMacStore,
		},
		transformer.PipelineStage{
			Name:        "JoinCnameAddressIdsWithMacAddresses",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(addressIdTableStore, cnameRecordTableStore)),
			Transformer: transformer.TransformFunc(joinAddressIdsWithMacAddresses),
			Writer:      cnameRecordsWithMacStore,
		},
		transformer.PipelineStage{
			Name:        "JoinARecordsWithCnameRecords",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(aRecordsWithMacStore, cnameRecordsWithMacStore)),
			Transformer: transformer.TransformFunc(joinARecordsWithCnameRecords),
			Writer:      allDnsMappingsStore,
		},
		transformer.PipelineStage{
			Name:        "EmitARecords",
			Reader:      excludeOldSessions(aRecordsWithMacStore),
			Transformer: transformer.MakeDoFunc(emitARecords),
			Writer:      allDnsMappingsStore,
		},
		transformer.PipelineStage{
			Name:        "JoinDomainsWithWhitelist",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(whitelistStore, allDnsMappingsStore)),
			Transformer: transformer.TransformFunc(joinDomainsWithWhitelist),
			Writer:      allWhitelistedMappingsStore,
		},
		transformer.PipelineStage{
			Name:        "JoinMacWithFlowId",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(addressIpTableStore, flowIpsTableStore)),
			Transformer: transformer.TransformFunc(joinMacWithFlowId),
			Writer:      flowMacsTableStore,
		},
		transformer.PipelineStage{
			Name:        "JoinWhitelistedDomainsWithFlows",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(allWhitelistedMappingsStore, flowMacsTableStore)),
			Transformer: transformer.TransformFunc(joinWhitelistedDomainsWithFlows),
			Writer:      flowDomainsTableStore,
		},
		transformer.PipelineStage{
			Name:        "GroupDomainsAndMacAddresses",
			Reader:      excludeOldSessions(flowDomainsTableStore),
			Transformer: transformer.TransformFunc(groupDomainsAndMacAddresses),
			Writer:      flowDomainsGroupedTableStore,
		},
		transformer.PipelineStage{
			Name:        "JoinDomainsWithSizes",
			Reader:      excludeOldSessions(store.NewDemuxingSeeker(flowDomainsGroupedTableStore, bytesPerTimestampShardedStore)),
			Transformer: transformer.TransformFunc(joinDomainsWithSizes),
			Writer:      bytesPerDomainShardedStore,
		},
		transformer.PipelineStage{
			Name:        "FlattenIntoBytesPerDevice",
			Reader:      bytesPerDomainShardedStore,
			Transformer: transformer.TransformFunc(flattenIntoBytesPerDevice),
			Writer:      bytesPerDomainPerDeviceStore,
		},
		transformer.PipelineStage{
			Name:        "FlattenIntoBytesPerTimestamp",
			Reader:      bytesPerDomainShardedStore,
			Transformer: transformer.TransformFunc(flattenIntoBytesPerTimestamp),
			Writer:      bytesPerDomainStore,
//.........这里部分代码省略.........

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:101，代码来源:bytesperdomain.go

示例19: SummarizeHealthPipeline

func SummarizeHealthPipeline(levelDbManager, csvManager store.Manager) transformer.Pipeline {
	memoryStore := levelDbManager.Reader("memory")
	memoryUsageByDayStore := levelDbManager.ReadingWriter("memory-usage-by-day")
	memoryUsageByDaySummarizedStore := levelDbManager.ReadingWriter("memory-usage-by-day-summarized")
	filesystemStore := levelDbManager.Reader("filesystem")
	filesystemUsageByDayStore := levelDbManager.ReadingWriter("filesystem-usage-by-day")
	filesystemUsageByDaySummarizedStore := levelDbManager.ReadingWriter("filesystem-usage-by-day-summarized")

	var timestamp, usage int64
	var filesystem, node string
	memoryUsageSummaryCsv := csvManager.Writer("memory-usage-summary.csv", []string{"timestamp", "node"}, []string{"usage"}, &timestamp, &node, &usage)
	filesystemUsageSummaryCsv := csvManager.Writer("filesystem-usage-summary.csv", []string{"filesystem", "timestamp", "node"}, []string{"usage"}, &filesystem, &timestamp, &node, &usage)

	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "OrderMemoryUsageByTimestamp",
			Reader:      memoryStore,
			Transformer: transformer.MakeMapFunc(orderRecordsByDay),
			Writer:      memoryUsageByDayStore,
		},
		transformer.PipelineStage{
			Name:        "SummarizeMemoryUsage",
			Reader:      memoryUsageByDayStore,
			Transformer: transformer.TransformFunc(summarizeMemoryUsage),
			Writer:      memoryUsageByDaySummarizedStore,
		},
		transformer.PipelineStage{
			Name:   "WriteMemoryUsageSummaryCsv",
			Reader: memoryUsageByDaySummarizedStore,
			Writer: memoryUsageSummaryCsv,
		},
		transformer.PipelineStage{
			Name:        "OrderFilesystemUsageByTimestamp",
			Reader:      filesystemStore,
			Transformer: transformer.MakeMapFunc(orderFilesystemRecordsByDay),
			Writer:      filesystemUsageByDayStore,
		},
		transformer.PipelineStage{
			Name:        "SummarizeFilesystemUsage",
			Reader:      filesystemUsageByDayStore,
			Transformer: transformer.TransformFunc(summarizeFilesystemUsage),
			Writer:      filesystemUsageByDaySummarizedStore,
		},
		transformer.PipelineStage{
			Name:   "WriteFilesystemUsageSummaryCsv",
			Reader: filesystemUsageByDaySummarizedStore,
			Writer: filesystemUsageSummaryCsv,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-tools，代码行数:50，代码来源:summarize.go

示例20: LookupsPerDevicePipeline

func LookupsPerDevicePipeline(levelDbManager store.Manager) transformer.Pipeline {
	tracesStore := levelDbManager.Seeker("traces")
	availabilityIntervalsStore := levelDbManager.Seeker("consistent-ranges")
	addressIdStore := levelDbManager.Seeker("bytesperdomain-address-id-table")
	addressIdToDomainStore := levelDbManager.SeekingWriter("lookupsperdevice-address-id-to-domain")
	lookupsPerDeviceSharded := levelDbManager.ReadingWriter("lookupsperdevice-sharded")
	lookupsPerDeviceStore := levelDbManager.Writer("lookupsperdevice-lookups-per-device")
	lookupsPerDevicePerHourStore := levelDbManager.Writer("lookupsperdevice-lookups-per-device-per-hour")
	consistentTracesStore := store.NewRangeIncludingReader(tracesStore, availabilityIntervalsStore)
	return []transformer.PipelineStage{
		transformer.PipelineStage{
			Name:        "LookupsPerDeviceMapper",
			Reader:      consistentTracesStore,
			Transformer: transformer.MakeDoFunc(lookupsPerDeviceMapper),
			Writer:      addressIdToDomainStore,
		},
		transformer.PipelineStage{
			Name:        "JoinMacWithLookups",
			Reader:      store.NewDemuxingSeeker(addressIdStore, addressIdToDomainStore),
			Transformer: transformer.TransformFunc(joinMacWithLookups),
			Writer:      lookupsPerDeviceSharded,
		},
		transformer.PipelineStage{
			Name:        "FlattenLookupsToNodeAndMac",
			Reader:      lookupsPerDeviceSharded,
			Transformer: transformer.TransformFunc(flattenLookupsToNodeAndMac),
			Writer:      lookupsPerDeviceStore,
		},
		transformer.PipelineStage{
			Name:        "FlattenLookupsToNodeMacAndTimestamp",
			Reader:      lookupsPerDeviceSharded,
			Transformer: transformer.TransformFunc(flattenLookupsToNodeMacAndTimestamp),
			Writer:      lookupsPerDevicePerHourStore,
		},
	}
}

开发者ID:sburnett，项目名称:bismark-passive-server-go，代码行数:36，代码来源:lookupsperdevice.go

注：本文中的github.com/sburnett/transformer/store.Manager类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。